evlli/hass-core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Refactor Teslemetry integration (#112480)

Browse source 
* Refactor Teslemetry

* Add abstractmethod

* Remove unused timestamp const

* Ruff

* Fix

* Update snapshots

* ruff

* Ruff

* ruff

* Lint

* Fix tests

* Fix tests and diag

* Refix snapshot

* Ruff

* Fix

* Fix bad merge

* has as property

* Remove _handle_coordinator_update

* Test and error changes
This commit is contained in:
Brett Adams 2024-05-10 18:52:33 +10:00 • committed by GitHub
parent 11f5b48724
commit 1a4e416bf4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 562 additions and 323 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
homeassistant/components/teslemetry/__init__.py
View file

@ -16,25 +16,30 @@ from homeassistant.const import CONF_ACCESS_TOKEN, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo
from .const import DOMAIN
from .const import DOMAIN, MODELS
from .coordinator import (
TeslemetryEnergyDataCoordinator,
TeslemetryEnergySiteLiveCoordinator,
TeslemetryVehicleDataCoordinator,
)
from .models import TeslemetryData, TeslemetryEnergyData, TeslemetryVehicleData
PLATFORMS: Final = [Platform.CLIMATE, Platform.SENSOR]
PLATFORMS: Final = [
Platform.CLIMATE,
Platform.SENSOR,
]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Teslemetry config."""
access_token = entry.data[CONF_ACCESS_TOKEN]
session = async_get_clientsession(hass)
# Create API connection
teslemetry = Teslemetry(
session=async_get_clientsession(hass),
session=session,
access_token=access_token,
)
try:
@ -52,36 +57,56 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
energysites: list[TeslemetryEnergyData] = []
for product in products:
if "vin" in product and Scope.VEHICLE_DEVICE_DATA in scopes:
# Remove the protobuff 'cached_data' that we do not use to save memory
product.pop("cached_data", None)
vin = product["vin"]
api = VehicleSpecific(teslemetry.vehicle, vin)
coordinator = TeslemetryVehicleDataCoordinator(hass, api)
coordinator = TeslemetryVehicleDataCoordinator(hass, api, product)
device = DeviceInfo(
identifiers={(DOMAIN, vin)},
manufacturer="Tesla",
configuration_url="https://teslemetry.com/console",
name=product["display_name"],
model=MODELS.get(vin[3]),
serial_number=vin,
)
vehicles.append(
TeslemetryVehicleData(
api=api,
coordinator=coordinator,
vin=vin,
device=device,
)
)
elif "energy_site_id" in product and Scope.ENERGY_DEVICE_DATA in scopes:
site_id = product["energy_site_id"]
api = EnergySpecific(teslemetry.energy, site_id)
live_coordinator = TeslemetryEnergySiteLiveCoordinator(hass, api)
device = DeviceInfo(
identifiers={(DOMAIN, str(site_id))},
manufacturer="Tesla",
configuration_url="https://teslemetry.com/console",
name=product.get("site_name", "Energy Site"),
)
energysites.append(
TeslemetryEnergyData(
api=api,
coordinator=TeslemetryEnergyDataCoordinator(hass, api),
live_coordinator=live_coordinator,
id=site_id,
info=product,
device=device,
)
)
# Do all coordinator first refreshes simultaneously
# Run all first refreshes
await asyncio.gather(
*(
vehicle.coordinator.async_config_entry_first_refresh()
for vehicle in vehicles
),
*(
energysite.coordinator.async_config_entry_first_refresh()
energysite.live_coordinator.async_config_entry_first_refresh()
for energysite in energysites
),
)

126
homeassistant/components/teslemetry/climate.py
View file

@ -2,11 +2,12 @@
from __future__ import annotations
from typing import Any
from typing import Any, cast
from tesla_fleet_api.const import Scope
from homeassistant.components.climate import (
ATTR_HVAC_MODE,
ClimateEntity,
ClimateEntityFeature,
HVACMode,
@ -17,10 +18,12 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .const import DOMAIN, TeslemetryClimateSide
from .context import handle_command
from .entity import TeslemetryVehicleEntity
from .models import TeslemetryVehicleData
DEFAULT_MIN_TEMP = 15
DEFAULT_MAX_TEMP = 28
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
@ -38,8 +41,7 @@ class TeslemetryClimateEntity(TeslemetryVehicleEntity, ClimateEntity):
"""Vehicle Location Climate Class."""
_attr_precision = PRECISION_HALVES
_attr_min_temp = 15
_attr_max_temp = 28
_attr_temperature_unit = UnitOfTemperature.CELSIUS
_attr_hvac_modes = [HVACMode.HEAT_COOL, HVACMode.OFF]
_attr_supported_features = (
@ -67,68 +69,65 @@ class TeslemetryClimateEntity(TeslemetryVehicleEntity, ClimateEntity):
side,
)
@property
def hvac_mode(self) -> HVACMode | None:
"""Return hvac operation ie. heat, cool mode."""
if self.get("climate_state_is_climate_on"):
return HVACMode.HEAT_COOL
return HVACMode.OFF
def _async_update_attrs(self) -> None:
"""Update the attributes of the entity."""
value = self.get("climate_state_is_climate_on")
if value is None:
self._attr_hvac_mode = None
elif value:
self._attr_hvac_mode = HVACMode.HEAT_COOL
else:
self._attr_hvac_mode = HVACMode.OFF
@property
def current_temperature(self) -> float | None:
"""Return the current temperature."""
return self.get("climate_state_inside_temp")
@property
def target_temperature(self) -> float | None:
"""Return the temperature we try to reach."""
return self.get(f"climate_state_{self.key}_setting")
@property
def max_temp(self) -> float:
"""Return the maximum temperature."""
return self.get("climate_state_max_avail_temp", self._attr_max_temp)
@property
def min_temp(self) -> float:
"""Return the minimum temperature."""
return self.get("climate_state_min_avail_temp", self._attr_min_temp)
@property
def preset_mode(self) -> str | None:
"""Return the current preset mode."""
return self.get("climate_state_climate_keeper_mode")
self._attr_current_temperature = self.get("climate_state_inside_temp")
self._attr_target_temperature = self.get(f"climate_state_{self.key}_setting")
self._attr_preset_mode = self.get("climate_state_climate_keeper_mode")
self._attr_min_temp = cast(
float, self.get("climate_state_min_avail_temp", DEFAULT_MIN_TEMP)
)
self._attr_max_temp = cast(
float, self.get("climate_state_max_avail_temp", DEFAULT_MAX_TEMP)
)
async def async_turn_on(self) -> None:
"""Set the climate state to on."""
self.raise_for_scope()
with handle_command():
await self.wake_up_if_asleep()
await self.api.auto_conditioning_start()
self.set(("climate_state_is_climate_on", True))
await self.wake_up_if_asleep()
await self.handle_command(self.api.auto_conditioning_start())
self._attr_hvac_mode = HVACMode.HEAT_COOL
self.async_write_ha_state()
async def async_turn_off(self) -> None:
"""Set the climate state to off."""
self.raise_for_scope()
with handle_command():
await self.wake_up_if_asleep()
await self.api.auto_conditioning_stop()
self.set(
("climate_state_is_climate_on", False),
("climate_state_climate_keeper_mode", "off"),
)
await self.wake_up_if_asleep()
await self.handle_command(self.api.auto_conditioning_stop())
self._attr_hvac_mode = HVACMode.OFF
self._attr_preset_mode = self._attr_preset_modes[0]
self.async_write_ha_state()
async def async_set_temperature(self, **kwargs: Any) -> None:
"""Set the climate temperature."""
temp = kwargs[ATTR_TEMPERATURE]
with handle_command():
await self.wake_up_if_asleep()
await self.api.set_temps(
driver_temp=temp,
passenger_temp=temp,
)
self.set((f"climate_state_{self.key}_setting", temp))
if temp := kwargs.get(ATTR_TEMPERATURE):
await self.wake_up_if_asleep()
await self.handle_command(
self.api.set_temps(
driver_temp=temp,
passenger_temp=temp,
)
)
self._attr_target_temperature = temp
if mode := kwargs.get(ATTR_HVAC_MODE):
# Set HVAC mode will call write_ha_state
await self.async_set_hvac_mode(mode)
else:
self.async_write_ha_state()
async def async_set_hvac_mode(self, hvac_mode: HVACMode) -> None:
"""Set the climate mode and state."""
@ -139,18 +138,15 @@ class TeslemetryClimateEntity(TeslemetryVehicleEntity, ClimateEntity):
async def async_set_preset_mode(self, preset_mode: str) -> None:
"""Set the climate preset mode."""
with handle_command():
await self.wake_up_if_asleep()
await self.api.set_climate_keeper_mode(
await self.wake_up_if_asleep()
await self.handle_command(
self.api.set_climate_keeper_mode(
climate_keeper_mode=self._attr_preset_modes.index(preset_mode)
)
self.set(
(
"climate_state_climate_keeper_mode",
preset_mode,
),
(
"climate_state_is_climate_on",
preset_mode != self._attr_preset_modes[0],
),
)
self._attr_preset_mode = preset_mode
if preset_mode == self._attr_preset_modes[0]:
self._attr_hvac_mode = HVACMode.OFF
else:
self._attr_hvac_mode = HVACMode.HEAT_COOL
self.async_write_ha_state()

8
homeassistant/components/teslemetry/const.py
View file

@ -10,10 +10,10 @@ DOMAIN = "teslemetry"
LOGGER = logging.getLogger(__package__)
MODELS = {
"model3": "Model 3",
"modelx": "Model X",
"modely": "Model Y",
"models": "Model S",
"S": "Model S",
"3": "Model 3",
"X": "Model X",
"Y": "Model Y",
}

16
homeassistant/components/teslemetry/context.py
View file

@ -1,16 +0,0 @@
"""Teslemetry context managers."""
from contextlib import contextmanager
from tesla_fleet_api.exceptions import TeslaFleetError
from homeassistant.exceptions import HomeAssistantError
@contextmanager
def handle_command():
"""Handle wake up and errors."""
try:
yield
except TeslaFleetError as e:
raise HomeAssistantError("Teslemetry command failed") from e

96
homeassistant/components/teslemetry/coordinator.py
View file

@ -13,12 +13,15 @@ from tesla_fleet_api.exceptions import (
)
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.exceptions import ConfigEntryAuthFailed
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import LOGGER, TeslemetryState
SYNC_INTERVAL = 60
VEHICLE_INTERVAL = timedelta(seconds=30)
ENERGY_LIVE_INTERVAL = timedelta(seconds=30)
ENERGY_INFO_INTERVAL = timedelta(seconds=30)
ENDPOINTS = [
VehicleDataEndpoint.CHARGE_STATE,
VehicleDataEndpoint.CLIMATE_STATE,
@ -29,50 +32,41 @@ ENDPOINTS = [
]
class TeslemetryDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Base class for Teslemetry Data Coordinators."""
def flatten(data: dict[str, Any], parent: str | None = None) -> dict[str, Any]:
"""Flatten the data structure."""
result = {}
for key, value in data.items():
if parent:
key = f"{parent}_{key}"
if isinstance(value, dict):
result.update(flatten(value, key))
else:
result[key] = value
return result
name: str
class TeslemetryVehicleDataCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Class to manage fetching data from the Teslemetry API."""
name = "Teslemetry Vehicle"
def __init__(
self, hass: HomeAssistant, api: VehicleSpecific | EnergySpecific
self, hass: HomeAssistant, api: VehicleSpecific, product: dict
) -> None:
"""Initialize Teslemetry Vehicle Update Coordinator."""
super().__init__(
hass,
LOGGER,
name=self.name,
update_interval=timedelta(seconds=SYNC_INTERVAL),
name="Teslemetry Vehicle",
update_interval=VEHICLE_INTERVAL,
)
self.api = api
class TeslemetryVehicleDataCoordinator(TeslemetryDataCoordinator):
"""Class to manage fetching data from the Teslemetry API."""
name = "Teslemetry Vehicle"
async def async_config_entry_first_refresh(self) -> None:
"""Perform first refresh."""
try:
response = await self.api.wake_up()
if response["response"]["state"] != TeslemetryState.ONLINE:
# The first refresh will fail, so retry later
raise ConfigEntryNotReady("Vehicle is not online")
except InvalidToken as e:
raise ConfigEntryAuthFailed from e
except SubscriptionRequired as e:
raise ConfigEntryAuthFailed from e
except TeslaFleetError as e:
# The first refresh will also fail, so retry later
raise ConfigEntryNotReady from e
await super().async_config_entry_first_refresh()
self.data = flatten(product)
async def _async_update_data(self) -> dict[str, Any]:
"""Update vehicle data using Teslemetry API."""
try:
data = await self.api.vehicle_data(endpoints=ENDPOINTS)
data = (await self.api.vehicle_data(endpoints=ENDPOINTS))["response"]
except VehicleOffline:
self.data["state"] = TeslemetryState.OFFLINE
return self.data
@ -83,33 +77,27 @@ class TeslemetryVehicleDataCoordinator(TeslemetryDataCoordinator):
except TeslaFleetError as e:
raise UpdateFailed(e.message) from e
return self._flatten(data["response"])
def _flatten(
self, data: dict[str, Any], parent: str | None = None
) -> dict[str, Any]:
"""Flatten the data structure."""
result = {}
for key, value in data.items():
if parent:
key = f"{parent}_{key}"
if isinstance(value, dict):
result.update(self._flatten(value, key))
else:
result[key] = value
return result
return flatten(data)
class TeslemetryEnergyDataCoordinator(TeslemetryDataCoordinator):
"""Class to manage fetching data from the Teslemetry API."""
class TeslemetryEnergySiteLiveCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Class to manage fetching energy site live status from the Teslemetry API."""
name = "Teslemetry Energy Site"
def __init__(self, hass: HomeAssistant, api: EnergySpecific) -> None:
"""Initialize Teslemetry Energy Site Live coordinator."""
super().__init__(
hass,
LOGGER,
name="Teslemetry Energy Site Live",
update_interval=ENERGY_LIVE_INTERVAL,
)
self.api = api
async def _async_update_data(self) -> dict[str, Any]:
"""Update energy site data using Teslemetry API."""
try:
data = await self.api.live_status()
data = (await self.api.live_status())["response"]
except InvalidToken as e:
raise ConfigEntryAuthFailed from e
except SubscriptionRequired as e:
@ -118,8 +106,8 @@ class TeslemetryEnergyDataCoordinator(TeslemetryDataCoordinator):
raise UpdateFailed(e.message) from e
# Convert Wall Connectors from array to dict
data["response"]["wall_connectors"] = {
wc["din"]: wc for wc in (data["response"].get("wall_connectors") or [])
data["wall_connectors"] = {
wc["din"]: wc for wc in (data.get("wall_connectors") or [])
}
return data["response"]
return data

3
homeassistant/components/teslemetry/diagnostics.py
View file

@ -36,7 +36,8 @@ async def async_get_config_entry_diagnostics(
x.coordinator.data for x in hass.data[DOMAIN][config_entry.entry_id].vehicles
]
energysites = [
x.coordinator.data for x in hass.data[DOMAIN][config_entry.entry_id].energysites
x.live_coordinator.data
for x in hass.data[DOMAIN][config_entry.entry_id].energysites
]
# Return only the relevant children

178
homeassistant/components/teslemetry/entity.py
View file

@ -1,52 +1,108 @@
"""Teslemetry parent entity class."""
from abc import abstractmethod
import asyncio
from typing import Any
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.exceptions import TeslaFleetError
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, MODELS, TeslemetryState
from .const import DOMAIN, LOGGER, TeslemetryState
from .coordinator import (
TeslemetryEnergyDataCoordinator,
TeslemetryEnergySiteLiveCoordinator,
TeslemetryVehicleDataCoordinator,
)
from .models import TeslemetryEnergyData, TeslemetryVehicleData
class TeslemetryVehicleEntity(CoordinatorEntity[TeslemetryVehicleDataCoordinator]):
"""Parent class for Teslemetry Vehicle Entities."""
class TeslemetryEntity(
CoordinatorEntity[
TeslemetryVehicleDataCoordinator | TeslemetryEnergySiteLiveCoordinator
]
):
"""Parent class for all Teslemetry entities."""
_attr_has_entity_name = True
def __init__(
self,
vehicle: TeslemetryVehicleData,
coordinator: TeslemetryVehicleDataCoordinator
| TeslemetryEnergySiteLiveCoordinator,
api: VehicleSpecific | EnergySpecific,
key: str,
) -> None:
"""Initialize common aspects of a Teslemetry entity."""
super().__init__(vehicle.coordinator)
super().__init__(coordinator)
self.api = api
self.key = key
self.api = vehicle.api
self._wakelock = vehicle.wakelock
self._attr_translation_key = self.key
self._async_update_attrs()
car_type = self.coordinator.data["vehicle_config_car_type"]
@property
def available(self) -> bool:
"""Return if sensor is available."""
return self.coordinator.last_update_success and self._attr_available
self._attr_translation_key = key
self._attr_unique_id = f"{vehicle.vin}-{key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, vehicle.vin)},
manufacturer="Tesla",
configuration_url="https://teslemetry.com/console",
name=self.coordinator.data["vehicle_state_vehicle_name"],
model=MODELS.get(car_type, car_type),
sw_version=self.coordinator.data["vehicle_state_car_version"].split(" ")[0],
hw_version=self.coordinator.data["vehicle_config_driver_assist"],
serial_number=vehicle.vin,
)
@property
def _value(self) -> Any | None:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(self.key)
def get(self, key: str, default: Any | None = None) -> Any | None:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(key, default)
@property
def is_none(self) -> bool:
"""Return if the value is a literal None."""
return self.get(self.key, False) is None
@property
def has(self) -> bool:
"""Return True if a specific value is in coordinator data."""
return self.key in self.coordinator.data
async def handle_command(self, command) -> dict[str, Any]:
"""Handle a command."""
try:
result = await command
LOGGER.debug("Command result: %s", result)
except TeslaFleetError as e:
LOGGER.debug("Command error: %s", e.message)
raise HomeAssistantError(f"Teslemetry command failed, {e.message}") from e
return result
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._async_update_attrs()
self.async_write_ha_state()
@abstractmethod
def _async_update_attrs(self) -> None:
"""Update the attributes of the entity."""
class TeslemetryVehicleEntity(TeslemetryEntity):
"""Parent class for Teslemetry Vehicle entities."""
_last_update: int = 0
def __init__(
self,
data: TeslemetryVehicleData,
key: str,
) -> None:
"""Initialize common aspects of a Teslemetry entity."""
self._attr_unique_id = f"{data.vin}-{key}"
self._wakelock = data.wakelock
self._attr_device_info = data.device
super().__init__(data.coordinator, data.api, key)
@property
def _value(self) -> Any | None:
@ -73,15 +129,27 @@ class TeslemetryVehicleEntity(CoordinatorEntity[TeslemetryVehicleDataCoordinator
raise HomeAssistantError("Could not wake up vehicle")
await asyncio.sleep(times * 5)
def get(self, key: str | None = None, default: Any | None = None) -> Any:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(key or self.key, default)
def set(self, *args: Any) -> None:
"""Set a value in coordinator data."""
for key, value in args:
self.coordinator.data[key] = value
self.async_write_ha_state()
async def handle_command(self, command) -> dict[str, Any]:
"""Handle a vehicle command."""
result = await super().handle_command(command)
if (response := result.get("response")) is None:
if message := result.get("error"):
# No response with error
LOGGER.info("Command failure: %s", message)
raise HomeAssistantError(message)
# No response without error (unexpected)
LOGGER.error("Unknown response: %s", response)
raise HomeAssistantError("Unknown response")
if (message := response.get("result")) is not True:
if message := response.get("reason"):
# Result of false with reason
LOGGER.info("Command failure: %s", message)
raise HomeAssistantError(message)
# Result of false without reason (unexpected)
LOGGER.error("Unknown response: %s", response)
raise HomeAssistantError("Unknown response")
# Response with result of true
return result
def raise_for_scope(self):
"""Raise an error if a scope is not available."""
@ -89,63 +157,53 @@ class TeslemetryVehicleEntity(CoordinatorEntity[TeslemetryVehicleDataCoordinator
raise ServiceValidationError("Missing required scope")
class TeslemetryEnergyEntity(CoordinatorEntity[TeslemetryEnergyDataCoordinator]):
"""Parent class for Teslemetry Energy Entities."""
_attr_has_entity_name = True
class TeslemetryEnergyLiveEntity(TeslemetryEntity):
"""Parent class for Teslemetry Energy Site Live entities."""
def __init__(
self,
energysite: TeslemetryEnergyData,
data: TeslemetryEnergyData,
key: str,
) -> None:
"""Initialize common aspects of a Teslemetry entity."""
super().__init__(energysite.coordinator)
self.key = key
self.api = energysite.api
"""Initialize common aspects of a Teslemetry Energy Site Live entity."""
self._attr_unique_id = f"{data.id}-{key}"
self._attr_device_info = data.device
self._attr_translation_key = key
self._attr_unique_id = f"{energysite.id}-{key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, str(energysite.id))},
manufacturer="Tesla",
configuration_url="https://teslemetry.com/console",
name=self.coordinator.data.get("site_name", "Energy Site"),
)
def get(self, key: str | None = None, default: Any | None = None) -> Any:
"""Return a specific value from coordinator data."""
return self.coordinator.data.get(key or self.key, default)
super().__init__(data.live_coordinator, data.api, key)
class TeslemetryWallConnectorEntity(CoordinatorEntity[TeslemetryEnergyDataCoordinator]):
class TeslemetryWallConnectorEntity(
TeslemetryEntity, CoordinatorEntity[TeslemetryEnergySiteLiveCoordinator]
):
"""Parent class for Teslemetry Wall Connector Entities."""
_attr_has_entity_name = True
def __init__(
self,
energysite: TeslemetryEnergyData,
data: TeslemetryEnergyData,
din: str,
key: str,
) -> None:
"""Initialize common aspects of a Teslemetry entity."""
super().__init__(energysite.coordinator)
self.din = din
self.key = key
self._attr_translation_key = key
self._attr_unique_id = f"{energysite.id}-{din}-{key}"
self._attr_unique_id = f"{data.id}-{din}-{key}"
self._attr_device_info = DeviceInfo(
identifiers={(DOMAIN, din)},
manufacturer="Tesla",
configuration_url="https://teslemetry.com/console",
name="Wall Connector",
via_device=(DOMAIN, str(energysite.id)),
via_device=(DOMAIN, str(data.id)),
serial_number=din.split("-")[-1],
)
super().__init__(data.live_coordinator, data.api, key)
@property
def _value(self) -> int:
"""Return a specific wall connector value from coordinator data."""
return self.coordinator.data["wall_connectors"][self.din].get(self.key)
return (
self.coordinator.data.get("wall_connectors", {})
.get(self.din, {})
.get(self.key)
)

9
homeassistant/components/teslemetry/models.py
View file

@ -8,8 +8,10 @@ from dataclasses import dataclass
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import Scope
from homeassistant.helpers.device_registry import DeviceInfo
from .coordinator import (
TeslemetryEnergyDataCoordinator,
TeslemetryEnergySiteLiveCoordinator,
TeslemetryVehicleDataCoordinator,
)
@ -31,6 +33,7 @@ class TeslemetryVehicleData:
coordinator: TeslemetryVehicleDataCoordinator
vin: str
wakelock = asyncio.Lock()
device: DeviceInfo
@dataclass
@ -38,6 +41,6 @@ class TeslemetryEnergyData:
"""Data for a vehicle in the Teslemetry integration."""
api: EnergySpecific
coordinator: TeslemetryEnergyDataCoordinator
live_coordinator: TeslemetryEnergySiteLiveCoordinator
id: int
info: dict[str, str]
device: DeviceInfo

72
homeassistant/components/teslemetry/sensor.py
View file

@ -4,7 +4,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime, timedelta
from datetime import timedelta
from itertools import chain
from typing import cast
@ -36,7 +36,7 @@ from homeassistant.util.variance import ignore_variance
from .const import DOMAIN
from .entity import (
TeslemetryEnergyEntity,
TeslemetryEnergyLiveEntity,
TeslemetryVehicleEntity,
TeslemetryWallConnectorEntity,
)
@ -298,7 +298,7 @@ VEHICLE_TIME_DESCRIPTIONS: tuple[TeslemetryTimeEntityDescription, ...] = (
),
)
ENERGY_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
ENERGY_LIVE_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
key="solar_power",
state_class=SensorStateClass.MEASUREMENT,
@ -421,15 +421,15 @@ async def async_setup_entry(
for description in VEHICLE_TIME_DESCRIPTIONS
),
( # Add energy site live
TeslemetryEnergySensorEntity(energysite, description)
TeslemetryEnergyLiveSensorEntity(energysite, description)
for energysite in data.energysites
for description in ENERGY_DESCRIPTIONS
if description.key in energysite.coordinator.data
for description in ENERGY_LIVE_DESCRIPTIONS
if description.key in energysite.live_coordinator.data
),
( # Add wall connectors
TeslemetryWallConnectorSensorEntity(energysite, din, description)
for energysite in data.energysites
for din in energysite.coordinator.data.get("wall_connectors", {})
for din in energysite.live_coordinator.data.get("wall_connectors", {})
for description in WALL_CONNECTOR_DESCRIPTIONS
),
)
@ -443,21 +443,23 @@ class TeslemetryVehicleSensorEntity(TeslemetryVehicleEntity, SensorEntity):
def __init__(
self,
vehicle: TeslemetryVehicleData,
data: TeslemetryVehicleData,
description: TeslemetrySensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(vehicle, description.key)
super().__init__(data, description.key)
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.entity_description.value_fn(self._value)
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
if self.has:
self._attr_native_value = self.entity_description.value_fn(self._value)
else:
self._attr_native_value = None
class TeslemetryVehicleTimeSensorEntity(TeslemetryVehicleEntity, SensorEntity):
"""Base class for Teslemetry vehicle metric sensors."""
"""Base class for Teslemetry vehicle time sensors."""
entity_description: TeslemetryTimeEntityDescription
@ -475,35 +477,31 @@ class TeslemetryVehicleTimeSensorEntity(TeslemetryVehicleEntity, SensorEntity):
super().__init__(data, description.key)
@property
def native_value(self) -> datetime | None:
"""Return the state of the sensor."""
return self._get_timestamp(self._value)
@property
def available(self) -> bool:
"""Return the availability of the sensor."""
return isinstance(self._value, int | float) and self._value > 0
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = isinstance(self._value, int | float) and self._value > 0
if self._attr_available:
self._attr_native_value = self._get_timestamp(self._value)
class TeslemetryEnergySensorEntity(TeslemetryEnergyEntity, SensorEntity):
class TeslemetryEnergyLiveSensorEntity(TeslemetryEnergyLiveEntity, SensorEntity):
"""Base class for Teslemetry energy site metric sensors."""
entity_description: SensorEntityDescription
def __init__(
self,
energysite: TeslemetryEnergyData,
data: TeslemetryEnergyData,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(energysite, description.key)
self.entity_description = description
super().__init__(data, description.key)
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self.get()
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = not self.is_none
self._attr_native_value = self._value
class TeslemetryWallConnectorSensorEntity(TeslemetryWallConnectorEntity, SensorEntity):
@ -513,19 +511,19 @@ class TeslemetryWallConnectorSensorEntity(TeslemetryWallConnectorEntity, SensorE
def __init__(
self,
energysite: TeslemetryEnergyData,
data: TeslemetryEnergyData,
din: str,
description: SensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
super().__init__(
energysite,
data,
din,
description.key,
)
self.entity_description = description
@property
def native_value(self) -> StateType:
"""Return the state of the sensor."""
return self._value
def _async_update_attrs(self) -> None:
"""Update the attributes of the sensor."""
self._attr_available = not self.is_none
self._attr_native_value = self._value

15
tests/components/teslemetry/conftest.py
View file

@ -8,10 +8,11 @@ from unittest.mock import patch
import pytest
from .const import (
COMMAND_OK,
LIVE_STATUS,
METADATA,
PRODUCTS,
RESPONSE_OK,
SITE_INFO,
VEHICLE_DATA,
WAKE_UP_ONLINE,
)
@ -70,7 +71,7 @@ def mock_request():
"""Mock Tesla Fleet API Vehicle Specific class."""
with patch(
"homeassistant.components.teslemetry.Teslemetry._request",
return_value=RESPONSE_OK,
return_value=COMMAND_OK,
) as mock_request:
yield mock_request
@ -83,3 +84,13 @@ def mock_live_status():
side_effect=lambda: deepcopy(LIVE_STATUS),
) as mock_live_status:
yield mock_live_status
@pytest.fixture(autouse=True)
def mock_site_info():
"""Mock Teslemetry Energy Specific site_info method."""
with patch(
"homeassistant.components.teslemetry.EnergySpecific.site_info",
side_effect=lambda: deepcopy(SITE_INFO),
) as mock_live_status:
yield mock_live_status

12
tests/components/teslemetry/const.py
View file

@ -14,6 +14,18 @@ PRODUCTS = load_json_object_fixture("products.json", DOMAIN)
VEHICLE_DATA = load_json_object_fixture("vehicle_data.json", DOMAIN)
VEHICLE_DATA_ALT = load_json_object_fixture("vehicle_data_alt.json", DOMAIN)
LIVE_STATUS = load_json_object_fixture("live_status.json", DOMAIN)
SITE_INFO = load_json_object_fixture("site_info.json", DOMAIN)
COMMAND_OK = {"response": {"result": True, "reason": ""}}
COMMAND_REASON = {"response": {"result": False, "reason": "already closed"}}
COMMAND_NOREASON = {"response": {"result": False}} # Unexpected
COMMAND_ERROR = {
"response": None,
"error": "vehicle unavailable: vehicle is offline or asleep",
"error_description": "",
}
COMMAND_NOERROR = {"answer": 42}
COMMAND_ERRORS = (COMMAND_REASON, COMMAND_NOREASON, COMMAND_ERROR, COMMAND_NOERROR)
RESPONSE_OK = {"response": {}, "error": None}

6
tests/components/teslemetry/fixtures/vehicle_data.json
View file

@ -73,14 +73,14 @@
},
"climate_state": {
"allow_cabin_overheat_protection": true,
"auto_seat_climate_left": false,
"auto_seat_climate_left": true,
"auto_seat_climate_right": true,
"auto_steering_wheel_heat": false,
"battery_heater": false,
"battery_heater_no_power": null,
"cabin_overheat_protection": "On",
"cabin_overheat_protection_actively_cooling": false,
"climate_keeper_mode": "off",
"climate_keeper_mode": "keep",
"cop_activation_temperature": "High",
"defrost_mode": 0,
"driver_temp_setting": 22,
@ -88,7 +88,7 @@
"hvac_auto_request": "On",
"inside_temp": 29.8,
"is_auto_conditioning_on": false,
"is_climate_on": false,
"is_climate_on": true,
"is_front_defroster_on": false,
"is_preconditioning": false,
"is_rear_defroster_on": false,

150
tests/components/teslemetry/snapshots/test_climate.ambr
View file

@ -46,6 +46,81 @@
})
# ---
# name: test_climate[climate.test_climate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 30.0,
'friendly_name': 'Test Climate',
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_mode': 'keep',
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
'supported_features': <ClimateEntityFeature: 401>,
'temperature': 22.0,
}),
'context': <ANY>,
'entity_id': 'climate.test_climate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'heat_cool',
})
# ---
# name: test_climate_alt[climate.test_climate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_climate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Climate',
'platform': 'teslemetry',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': <TeslemetryClimateSide.DRIVER: 'driver_temp'>,
'unique_id': 'VINVINVIN-driver_temp',
'unit_of_measurement': None,
})
# ---
# name: test_climate_alt[climate.test_climate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': 30.0,
@ -74,3 +149,78 @@
'state': 'off',
})
# ---
# name: test_climate_offline[climate.test_climate-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
}),
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'climate',
'entity_category': None,
'entity_id': 'climate.test_climate',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Climate',
'platform': 'teslemetry',
'previous_unique_id': None,
'supported_features': <ClimateEntityFeature: 401>,
'translation_key': <TeslemetryClimateSide.DRIVER: 'driver_temp'>,
'unique_id': 'VINVINVIN-driver_temp',
'unit_of_measurement': None,
})
# ---
# name: test_climate_offline[climate.test_climate-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'current_temperature': None,
'friendly_name': 'Test Climate',
'hvac_modes': list([
<HVACMode.HEAT_COOL: 'heat_cool'>,
<HVACMode.OFF: 'off'>,
]),
'max_temp': 28.0,
'min_temp': 15.0,
'preset_mode': None,
'preset_modes': list([
'off',
'keep',
'dog',
'camp',
]),
'supported_features': <ClimateEntityFeature: 401>,
'temperature': None,
}),
'context': <ANY>,
'entity_id': 'climate.test_climate',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---

6
tests/components/teslemetry/snapshots/test_diagnostics.ambr
View file

@ -94,14 +94,14 @@
'charge_state_usable_battery_level': 77,
'charge_state_user_charge_enable_request': None,
'climate_state_allow_cabin_overheat_protection': True,
'climate_state_auto_seat_climate_left': False,
'climate_state_auto_seat_climate_left': True,
'climate_state_auto_seat_climate_right': True,
'climate_state_auto_steering_wheel_heat': False,
'climate_state_battery_heater': False,
'climate_state_battery_heater_no_power': None,
'climate_state_cabin_overheat_protection': 'On',
'climate_state_cabin_overheat_protection_actively_cooling': False,
'climate_state_climate_keeper_mode': 'off',
'climate_state_climate_keeper_mode': 'keep',
'climate_state_cop_activation_temperature': 'High',
'climate_state_defrost_mode': 0,
'climate_state_driver_temp_setting': 22,
@ -109,7 +109,7 @@
'climate_state_hvac_auto_request': 'On',
'climate_state_inside_temp': 29.8,
'climate_state_is_auto_conditioning_on': False,
'climate_state_is_climate_on': False,
'climate_state_is_climate_on': True,
'climate_state_is_front_defroster_on': False,
'climate_state_is_preconditioning': False,
'climate_state_is_rear_defroster_on': False,

84
tests/components/teslemetry/test_climate.py
View file

@ -1,6 +1,5 @@
"""Test the Teslemetry climate platform."""
from datetime import timedelta
from unittest.mock import patch
from freezegun.api import FrozenDateTimeFactory
@ -19,14 +18,20 @@ from homeassistant.components.climate import (
SERVICE_TURN_ON,
HVACMode,
)
from homeassistant.components.teslemetry.coordinator import SYNC_INTERVAL
from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL
from homeassistant.const import ATTR_ENTITY_ID, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError, ServiceValidationError
from homeassistant.helpers import entity_registry as er
from . import assert_entities, setup_platform
from .const import METADATA_NOSCOPE, WAKE_UP_ASLEEP, WAKE_UP_ONLINE
from .const import (
COMMAND_ERRORS,
METADATA_NOSCOPE,
VEHICLE_DATA_ALT,
WAKE_UP_ASLEEP,
WAKE_UP_ONLINE,
)
from tests.common import async_fire_time_changed
@ -43,27 +48,34 @@ async def test_climate(
assert_entities(hass, entry.entry_id, entity_registry, snapshot)
entity_id = "climate.test_climate"
state = hass.states.get(entity_id)
# Turn On
# Turn On and Set Temp
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_HVAC_MODE,
{ATTR_ENTITY_ID: [entity_id], ATTR_HVAC_MODE: HVACMode.HEAT_COOL},
SERVICE_SET_TEMPERATURE,
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 20,
ATTR_HVAC_MODE: HVACMode.HEAT_COOL,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 20
assert state.state == HVACMode.HEAT_COOL
# Set Temp
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_TEMPERATURE,
{ATTR_ENTITY_ID: [entity_id], ATTR_TEMPERATURE: 20},
{
ATTR_ENTITY_ID: [entity_id],
ATTR_TEMPERATURE: 21,
},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_TEMPERATURE] == 20
assert state.attributes[ATTR_TEMPERATURE] == 21
# Set Preset
await hass.services.async_call(
@ -75,6 +87,16 @@ async def test_climate(
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == "keep"
# Set Preset
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_SET_PRESET_MODE,
{ATTR_ENTITY_ID: [entity_id], ATTR_PRESET_MODE: "off"},
blocking=True,
)
state = hass.states.get(entity_id)
assert state.attributes[ATTR_PRESET_MODE] == "off"
# Turn Off
await hass.services.async_call(
CLIMATE_DOMAIN,
@ -86,9 +108,34 @@ async def test_climate(
assert state.state == HVACMode.OFF
async def test_errors(
async def test_climate_alt(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data,
) -> None:
"""Tests that the climate entity is correct."""
mock_vehicle_data.return_value = VEHICLE_DATA_ALT
entry = await setup_platform(hass, [Platform.CLIMATE])
assert_entities(hass, entry.entry_id, entity_registry, snapshot)
async def test_climate_offline(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data,
) -> None:
"""Tests that the climate entity is correct."""
mock_vehicle_data.side_effect = VehicleOffline
entry = await setup_platform(hass, [Platform.CLIMATE])
assert_entities(hass, entry.entry_id, entity_registry, snapshot)
@pytest.mark.parametrize("response", COMMAND_ERRORS)
async def test_errors(hass: HomeAssistant, response: str) -> None:
"""Tests service error is handled."""
await setup_platform(hass, platforms=[Platform.CLIMATE])
@ -110,6 +157,21 @@ async def test_errors(
mock_on.assert_called_once()
assert error.from_exception == InvalidCommand
with (
patch(
"homeassistant.components.teslemetry.VehicleSpecific.auto_conditioning_start",
return_value=response,
) as mock_on,
pytest.raises(HomeAssistantError) as error,
):
await hass.services.async_call(
CLIMATE_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: [entity_id]},
blocking=True,
)
mock_on.assert_called_once()
async def test_asleep_or_offline(
hass: HomeAssistant,
@ -127,7 +189,7 @@ async def test_asleep_or_offline(
# Put the vehicle alseep
mock_vehicle_data.reset_mock()
mock_vehicle_data.side_effect = VehicleOffline
freezer.tick(timedelta(seconds=SYNC_INTERVAL))
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_vehicle_data.assert_called_once()

55
tests/components/teslemetry/test_init.py
View file

@ -1,7 +1,5 @@
"""Test the Tessie init."""
from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory
import pytest
from tesla_fleet_api.exceptions import (
@ -11,13 +9,12 @@ from tesla_fleet_api.exceptions import (
VehicleOffline,
)
from homeassistant.components.teslemetry.coordinator import SYNC_INTERVAL
from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL
from homeassistant.config_entries import ConfigEntryState
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from . import setup_platform
from .const import WAKE_UP_ASLEEP, WAKE_UP_ONLINE
from tests.common import async_fire_time_changed
@ -50,48 +47,6 @@ async def test_init_error(
# Vehicle Coordinator
async def test_vehicle_first_refresh(
hass: HomeAssistant,
mock_wake_up,
mock_vehicle_data,
mock_products,
freezer: FrozenDateTimeFactory,
) -> None:
"""Test first coordinator refresh but vehicle is asleep."""
# Mock vehicle is asleep
mock_wake_up.return_value = WAKE_UP_ASLEEP
entry = await setup_platform(hass)
assert entry.state is ConfigEntryState.SETUP_RETRY
mock_wake_up.assert_called_once()
# Reset mock and set vehicle to online
mock_wake_up.reset_mock()
mock_wake_up.return_value = WAKE_UP_ONLINE
# Wait for the retry
freezer.tick(timedelta(seconds=60))
async_fire_time_changed(hass)
await hass.async_block_till_done(wait_background_tasks=True)
# Verify we have loaded
assert entry.state is ConfigEntryState.LOADED
mock_wake_up.assert_called_once()
mock_vehicle_data.assert_called_once()
@pytest.mark.parametrize(("side_effect", "state"), ERRORS)
async def test_vehicle_first_refresh_error(
hass: HomeAssistant, mock_wake_up, side_effect, state
) -> None:
"""Test first coordinator refresh with an error."""
mock_wake_up.side_effect = side_effect
entry = await setup_platform(hass)
assert entry.state is state
async def test_vehicle_refresh_offline(
hass: HomeAssistant, mock_vehicle_data, freezer: FrozenDateTimeFactory
) -> None:
@ -102,7 +57,7 @@ async def test_vehicle_refresh_offline(
mock_vehicle_data.reset_mock()
mock_vehicle_data.side_effect = VehicleOffline
freezer.tick(timedelta(seconds=SYNC_INTERVAL))
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()
mock_vehicle_data.assert_called_once()
@ -118,11 +73,9 @@ async def test_vehicle_refresh_error(
assert entry.state is state
# Test Energy Coordinator
# Test Energy Live Coordinator
@pytest.mark.parametrize(("side_effect", "state"), ERRORS)
async def test_energy_refresh_error(
async def test_energy_live_refresh_error(
hass: HomeAssistant, mock_live_status, side_effect, state
) -> None:
"""Test coordinator refresh with an error."""

6
tests/components/teslemetry/test_sensor.py
View file

@ -1,12 +1,10 @@
"""Test the Teslemetry sensor platform."""
from datetime import timedelta
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy import SnapshotAssertion
from homeassistant.components.teslemetry.coordinator import SYNC_INTERVAL
from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import entity_registry as er
@ -35,7 +33,7 @@ async def test_sensors(
# Coordinator refresh
mock_vehicle_data.return_value = VEHICLE_DATA_ALT
freezer.tick(timedelta(seconds=SYNC_INTERVAL))
freezer.tick(VEHICLE_INTERVAL)
async_fire_time_changed(hass)
await hass.async_block_till_done()