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

Add streaming sensors to Teslemetry (#132783) 

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
pull/135224/head
Brett Adams 2025-01-10 03:58:12 +10:00 committed by GitHub
parent cabdae98e8
commit b6c0257c43
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 497 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
homeassistant/components/teslemetry/__init__.py
View File

@ -126,13 +126,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
create_handle_vehicle_stream(vin, coordinator),
{"vin": vin},
)
firmware = vehicle_metadata[vin].get("firmware", "Unknown")
vehicles.append(
TeslemetryVehicleData(
api=api,
config_entry=entry,
coordinator=coordinator,
stream=stream,
vin=vin,
firmware=firmware,
device=device,
remove_listener=remove_listener,
)
@ -179,6 +182,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
# Run all first refreshes
await asyncio.gather(
*(async_setup_stream(hass, entry, vehicle) for vehicle in vehicles),
*(
vehicle.coordinator.async_config_entry_first_refresh()
for vehicle in vehicles
@ -265,3 +269,15 @@ def create_handle_vehicle_stream(vin: str, coordinator) -> Callable[[dict], None
coordinator.async_set_updated_data(coordinator.data)
return handle_vehicle_stream
async def async_setup_stream(
hass: HomeAssistant, entry: ConfigEntry, vehicle: TeslemetryVehicleData
):
"""Set up the stream for a vehicle."""
vehicle_stream = vehicle.stream.get_vehicle(vehicle.vin)
await vehicle_stream.get_config()
entry.async_create_background_task(
hass, vehicle_stream.prefer_typed(True), f"Prefer typed for {vehicle.vin}"
)

53
homeassistant/components/teslemetry/entity.py
View File

@ -5,9 +5,11 @@ from typing import Any
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import Scope
from teslemetry_stream import Signal
from homeassistant.exceptions import ServiceValidationError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
@ -73,11 +75,6 @@ class TeslemetryEntity(
"""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
def _handle_coordinator_update(self) -> None:
"""Handle updated data from the coordinator."""
self._async_update_attrs()
@ -236,3 +233,49 @@ class TeslemetryWallConnectorEntity(TeslemetryEntity):
return self.key in self.coordinator.data.get("wall_connectors", {}).get(
self.din, {}
)
class TeslemetryVehicleStreamEntity(Entity):
"""Parent class for Teslemetry Vehicle Stream entities."""
_attr_has_entity_name = True
def __init__(
self, data: TeslemetryVehicleData, key: str, streaming_key: Signal
) -> None:
"""Initialize common aspects of a Teslemetry entity."""
self.streaming_key = streaming_key
self.vehicle = data
self.api = data.api
self.stream = data.stream
self.vin = data.vin
self.add_field = data.stream.get_vehicle(self.vin).add_field
self._attr_translation_key = key
self._attr_unique_id = f"{data.vin}-{key}"
self._attr_device_info = data.device
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self.async_on_remove(
self.stream.async_add_listener(
self._handle_stream_update,
{"vin": self.vin, "data": {self.streaming_key: None}},
)
)
self.vehicle.config_entry.async_create_background_task(
self.hass,
self.add_field(self.streaming_key),
f"Adding field {self.streaming_key.value} to {self.vehicle.vin}",
)
def _handle_stream_update(self, data: dict[str, Any]) -> None:
"""Handle updated data from the stream."""
self._async_value_from_stream(data["data"][self.streaming_key])
self.async_write_ha_state()
def _async_value_from_stream(self, value: Any) -> None:
"""Update the entity with the latest value from the stream."""
raise NotImplementedError

2
homeassistant/components/teslemetry/manifest.json
View File

@ -6,5 +6,5 @@
"documentation": "https://www.home-assistant.io/integrations/teslemetry",
"iot_class": "cloud_polling",
"loggers": ["tesla-fleet-api"],
"requirements": ["tesla-fleet-api==0.9.2", "teslemetry-stream==0.4.2"]
"requirements": ["tesla-fleet-api==0.9.2", "teslemetry-stream==0.5.3"]
}

5
homeassistant/components/teslemetry/models.py
View File

@ -10,6 +10,7 @@ from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import Scope
from teslemetry_stream import TeslemetryStream
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceInfo
from .coordinator import (
@ -34,12 +35,14 @@ class TeslemetryVehicleData:
"""Data for a vehicle in the Teslemetry integration."""
api: VehicleSpecific
config_entry: ConfigEntry
coordinator: TeslemetryVehicleDataCoordinator
stream: TeslemetryStream
vin: str
wakelock = asyncio.Lock()
firmware: str
device: DeviceInfo
remove_listener: Callable
wakelock = asyncio.Lock()
@dataclass

318
homeassistant/components/teslemetry/sensor.py
View File

@ -5,10 +5,12 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from datetime import timedelta
from itertools import chain
from typing import cast
from propcache import cached_property
from teslemetry_stream import Signal
from homeassistant.components.sensor import (
RestoreSensor,
SensorDeviceClass,
SensorEntity,
SensorEntityDescription,
@ -40,6 +42,7 @@ from .entity import (
TeslemetryEnergyInfoEntity,
TeslemetryEnergyLiveEntity,
TeslemetryVehicleEntity,
TeslemetryVehicleStreamEntity,
TeslemetryWallConnectorEntity,
)
from .models import TeslemetryEnergyData, TeslemetryVehicleData
@ -59,125 +62,165 @@ SHIFT_STATES = {"P": "p", "D": "d", "R": "r", "N": "n"}
@dataclass(frozen=True, kw_only=True)
class TeslemetrySensorEntityDescription(SensorEntityDescription):
class TeslemetryVehicleSensorEntityDescription(SensorEntityDescription):
"""Describes Teslemetry Sensor entity."""
value_fn: Callable[[StateType], StateType] = lambda x: x
polling: bool = False
polling_value_fn: Callable[[StateType], StateType] = lambda x: x
polling_available_fn: Callable[[StateType], bool] = lambda x: x is not None
streaming_key: Signal | None = None
streaming_value_fn: Callable[[StateType], StateType] = lambda x: x
streaming_firmware: str = "2024.26"
VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
TeslemetrySensorEntityDescription(
VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charging_state",
polling=True,
streaming_key=Signal.DETAILED_CHARGE_STATE,
polling_value_fn=lambda value: CHARGE_STATES.get(str(value)),
streaming_value_fn=lambda value: CHARGE_STATES.get(
str(value).replace("DetailedChargeState", "")
),
options=list(CHARGE_STATES.values()),
device_class=SensorDeviceClass.ENUM,
value_fn=lambda value: CHARGE_STATES.get(cast(str, value)),
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_battery_level",
polling=True,
streaming_key=Signal.BATTERY_LEVEL,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
suggested_display_precision=1,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_usable_battery_level",
polling=True,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charge_energy_added",
polling=True,
streaming_key=Signal.AC_CHARGING_ENERGY_IN,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
suggested_display_precision=1,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charger_power",
polling=True,
streaming_key=Signal.AC_CHARGING_POWER,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.KILO_WATT,
device_class=SensorDeviceClass.POWER,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charger_voltage",
polling=True,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
device_class=SensorDeviceClass.VOLTAGE,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charger_actual_current",
polling=True,
streaming_key=Signal.CHARGE_AMPS,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
device_class=SensorDeviceClass.CURRENT,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charge_rate",
polling=True,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
device_class=SensorDeviceClass.SPEED,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_conn_charge_cable",
polling=True,
streaming_key=Signal.CHARGING_CABLE_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_fast_charger_type",
polling=True,
streaming_key=Signal.FAST_CHARGER_TYPE,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_battery_range",
polling=True,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=1,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_est_battery_range",
polling=True,
streaming_key=Signal.EST_BATTERY_RANGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=1,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="charge_state_ideal_battery_range",
polling=True,
streaming_key=Signal.IDEAL_BATTERY_RANGE,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
suggested_display_precision=1,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="drive_state_speed",
polling=True,
polling_value_fn=lambda value: value or 0,
streaming_key=Signal.VEHICLE_SPEED,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
device_class=SensorDeviceClass.SPEED,
entity_registry_enabled_default=False,
value_fn=lambda value: value or 0,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="drive_state_power",
polling=True,
polling_value_fn=lambda value: value or 0,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.KILO_WATT,
device_class=SensorDeviceClass.POWER,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
value_fn=lambda value: value or 0,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="drive_state_shift_state",
polling=True,
polling_available_fn=lambda x: True,
polling_value_fn=lambda x: SHIFT_STATES.get(str(x), "p"),
streaming_key=Signal.GEAR,
streaming_value_fn=lambda x: SHIFT_STATES.get(str(x)),
options=list(SHIFT_STATES.values()),
device_class=SensorDeviceClass.ENUM,
value_fn=lambda x: SHIFT_STATES.get(str(x), "p"),
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_odometer",
polling=True,
streaming_key=Signal.ODOMETER,
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
@ -185,8 +228,10 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_fl",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_FL,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -195,8 +240,10 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_fr",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_FR,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -205,8 +252,10 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_rl",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_RL,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -215,8 +264,10 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_rr",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_RR,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -225,22 +276,27 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="climate_state_inside_temp",
polling=True,
streaming_key=Signal.INSIDE_TEMP,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
suggested_display_precision=1,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="climate_state_outside_temp",
polling=True,
streaming_key=Signal.OUTSIDE_TEMP,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
suggested_display_precision=1,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="climate_state_driver_temp_setting",
polling=True,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
@ -248,8 +304,9 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="climate_state_passenger_temp_setting",
polling=True,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
@ -257,23 +314,29 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="drive_state_active_route_traffic_minutes_delay",
polling=True,
streaming_key=Signal.ROUTE_TRAFFIC_MINUTES_DELAY,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=SensorDeviceClass.DURATION,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="drive_state_active_route_energy_at_arrival",
polling=True,
streaming_key=Signal.EXPECTED_ENERGY_PERCENT_AT_TRIP_ARRIVAL,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetrySensorEntityDescription(
TeslemetryVehicleSensorEntityDescription(
key="drive_state_active_route_miles_to_arrival",
polling=True,
streaming_key=Signal.MILES_TO_ARRIVAL,
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
@ -286,17 +349,21 @@ class TeslemetryTimeEntityDescription(SensorEntityDescription):
"""Describes Teslemetry Sensor entity."""
variance: int
streaming_key: Signal
streaming_firmware: str = "2024.26"
VEHICLE_TIME_DESCRIPTIONS: tuple[TeslemetryTimeEntityDescription, ...] = (
TeslemetryTimeEntityDescription(
key="charge_state_minutes_to_full_charge",
streaming_key=Signal.TIME_TO_FULL_CHARGE,
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
variance=4,
),
TeslemetryTimeEntityDescription(
key="drive_state_active_route_minutes_to_arrival",
streaming_key=Signal.MINUTES_TO_ARRIVAL,
device_class=SensorDeviceClass.TIMESTAMP,
variance=1,
),
@ -391,6 +458,14 @@ ENERGY_LIVE_DESCRIPTIONS: tuple[SensorEntityDescription, ...] = (
),
)
@dataclass(frozen=True, kw_only=True)
class TeslemetrySensorEntityDescription(SensorEntityDescription):
"""Describes Teslemetry Sensor entity."""
value_fn: Callable[[StateType], StateType] = lambda x: x
WALL_CONNECTOR_DESCRIPTIONS: tuple[TeslemetrySensorEntityDescription, ...] = (
TeslemetrySensorEntityDescription(
key="wall_connector_state",
@ -448,55 +523,106 @@ async def async_setup_entry(
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Teslemetry sensor platform from a config entry."""
async_add_entities(
chain(
( # Add vehicles
TeslemetryVehicleSensorEntity(vehicle, description)
for vehicle in entry.runtime_data.vehicles
for description in VEHICLE_DESCRIPTIONS
),
( # Add vehicles time sensors
TeslemetryVehicleTimeSensorEntity(vehicle, description)
for vehicle in entry.runtime_data.vehicles
for description in VEHICLE_TIME_DESCRIPTIONS
),
( # Add energy site live
TeslemetryEnergyLiveSensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_LIVE_DESCRIPTIONS
if description.key in energysite.live_coordinator.data
),
( # Add wall connectors
TeslemetryWallConnectorSensorEntity(energysite, din, description)
for energysite in entry.runtime_data.energysites
for din in energysite.live_coordinator.data.get("wall_connectors", {})
for description in WALL_CONNECTOR_DESCRIPTIONS
),
( # Add energy site info
TeslemetryEnergyInfoSensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_INFO_DESCRIPTIONS
if description.key in energysite.info_coordinator.data
),
( # Add energy history sensor
TeslemetryEnergyHistorySensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_HISTORY_DESCRIPTIONS
if energysite.history_coordinator
),
)
entities: list[SensorEntity] = []
for vehicle in entry.runtime_data.vehicles:
for description in VEHICLE_DESCRIPTIONS:
if (
not vehicle.api.pre2021
and description.streaming_key
and vehicle.firmware >= description.streaming_firmware
):
entities.append(TeslemetryStreamSensorEntity(vehicle, description))
elif description.polling:
entities.append(TeslemetryVehicleSensorEntity(vehicle, description))
for time_description in VEHICLE_TIME_DESCRIPTIONS:
if (
not vehicle.api.pre2021
and vehicle.firmware >= time_description.streaming_firmware
):
entities.append(
TeslemetryStreamTimeSensorEntity(vehicle, time_description)
)
else:
entities.append(
TeslemetryVehicleTimeSensorEntity(vehicle, time_description)
)
entities.extend(
TeslemetryEnergyLiveSensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_LIVE_DESCRIPTIONS
if description.key in energysite.live_coordinator.data
)
entities.extend(
TeslemetryWallConnectorSensorEntity(energysite, din, description)
for energysite in entry.runtime_data.energysites
for din in energysite.live_coordinator.data.get("wall_connectors", {})
for description in WALL_CONNECTOR_DESCRIPTIONS
)
entities.extend(
TeslemetryEnergyInfoSensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_INFO_DESCRIPTIONS
if description.key in energysite.info_coordinator.data
)
entities.extend(
TeslemetryEnergyHistorySensorEntity(energysite, description)
for energysite in entry.runtime_data.energysites
for description in ENERGY_HISTORY_DESCRIPTIONS
if energysite.history_coordinator is not None
)
async_add_entities(entities)
class TeslemetryStreamSensorEntity(TeslemetryVehicleStreamEntity, RestoreSensor):
"""Base class for Teslemetry vehicle streaming sensors."""
entity_description: TeslemetryVehicleSensorEntityDescription
def __init__(
self,
data: TeslemetryVehicleData,
description: TeslemetryVehicleSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
assert description.streaming_key
super().__init__(data, description.key, description.streaming_key)
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if (sensor_data := await self.async_get_last_sensor_data()) is not None:
self._attr_native_value = sensor_data.native_value
@cached_property
def available(self) -> bool:
"""Return True if entity is available."""
return self.stream.connected
def _async_value_from_stream(self, value) -> None:
"""Update the value of the entity."""
if value is None:
self._attr_native_value = None
else:
self._attr_native_value = self.entity_description.streaming_value_fn(value)
class TeslemetryVehicleSensorEntity(TeslemetryVehicleEntity, SensorEntity):
"""Base class for Teslemetry vehicle metric sensors."""
entity_description: TeslemetrySensorEntityDescription
entity_description: TeslemetryVehicleSensorEntityDescription
def __init__(
self,
data: TeslemetryVehicleData,
description: TeslemetrySensorEntityDescription,
description: TeslemetryVehicleSensorEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
@ -504,12 +630,48 @@ class TeslemetryVehicleSensorEntity(TeslemetryVehicleEntity, SensorEntity):
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)
if self.entity_description.polling_available_fn(self._value):
self._attr_available = True
self._attr_native_value = self.entity_description.polling_value_fn(
self._value
)
else:
self._attr_available = False
self._attr_native_value = None
class TeslemetryStreamTimeSensorEntity(TeslemetryVehicleStreamEntity, SensorEntity):
"""Base class for Teslemetry vehicle streaming sensors."""
entity_description: TeslemetryTimeEntityDescription
def __init__(
self,
data: TeslemetryVehicleData,
description: TeslemetryTimeEntityDescription,
) -> None:
"""Initialize the sensor."""
self.entity_description = description
self._get_timestamp = ignore_variance(
func=lambda value: dt_util.now() + timedelta(minutes=value),
ignored_variance=timedelta(minutes=description.variance),
)
assert description.streaming_key
super().__init__(data, description.key, description.streaming_key)
@cached_property
def available(self) -> bool:
"""Return True if entity is available."""
return self.stream.connected
def _async_value_from_stream(self, value) -> None:
"""Update the value of the entity."""
if value is None:
self._attr_native_value = None
else:
self._attr_native_value = self._get_timestamp(value)
class TeslemetryVehicleTimeSensorEntity(TeslemetryVehicleEntity, SensorEntity):
"""Base class for Teslemetry vehicle time sensors."""

2
requirements_all.txt generated
View File

@ -2853,7 +2853,7 @@ tesla-powerwall==0.5.2
tesla-wall-connector==1.0.2
# homeassistant.components.teslemetry
teslemetry-stream==0.4.2
teslemetry-stream==0.5.3
# homeassistant.components.tessie
tessie-api==0.1.1

2
requirements_test_all.txt generated
View File

@ -2290,7 +2290,7 @@ tesla-powerwall==0.5.2
tesla-wall-connector==1.0.2
# homeassistant.components.teslemetry
teslemetry-stream==0.4.2
teslemetry-stream==0.5.3
# homeassistant.components.tessie
tessie-api==0.1.1

53
tests/components/teslemetry/conftest.py
View File

@ -7,6 +7,7 @@ from copy import deepcopy
from unittest.mock import AsyncMock, patch
import pytest
from teslemetry_stream.stream import recursive_match
from .const import (
COMMAND_OK,
@ -109,9 +110,53 @@ def mock_energy_history():
@pytest.fixture(autouse=True)
def mock_listen():
def mock_add_listener():
"""Mock Teslemetry Stream listen method."""
with patch(
"homeassistant.components.teslemetry.TeslemetryStream.listen",
) as mock_listen:
yield mock_listen
"homeassistant.components.teslemetry.TeslemetryStream.async_add_listener",
) as mock_add_listener:
mock_add_listener.listeners = []
def unsubscribe() -> None:
return
def side_effect(callback, filters):
mock_add_listener.listeners.append((callback, filters))
return unsubscribe
def send(event) -> None:
for listener, filters in mock_add_listener.listeners:
if recursive_match(filters, event):
listener(event)
mock_add_listener.send = send
mock_add_listener.side_effect = side_effect
yield mock_add_listener
@pytest.fixture(autouse=True)
def mock_stream_get_config():
"""Mock Teslemetry Stream listen method."""
with patch(
"teslemetry_stream.TeslemetryStreamVehicle.get_config",
) as mock_stream_get_config:
yield mock_stream_get_config
@pytest.fixture(autouse=True)
def mock_stream_update_config():
"""Mock Teslemetry Stream listen method."""
with patch(
"teslemetry_stream.TeslemetryStreamVehicle.update_config",
) as mock_stream_update_config:
yield mock_stream_update_config
@pytest.fixture(autouse=True)
def mock_stream_connected():
"""Mock Teslemetry Stream listen method."""
with patch(
"homeassistant.components.teslemetry.TeslemetryStream.connected",
return_value=True,
) as mock_stream_connected:
yield mock_stream_connected

1
tests/components/teslemetry/const.py
View File

@ -18,6 +18,7 @@ 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)
ENERGY_HISTORY = load_json_object_fixture("energy_history.json", DOMAIN)
METADATA = load_json_object_fixture("metadata.json", DOMAIN)
COMMAND_OK = {"response": {"result": True, "reason": ""}}
COMMAND_REASON = {"response": {"result": False, "reason": "already closed"}}

10
tests/components/teslemetry/fixtures/config.json Normal file
View File

@ -0,0 +1,10 @@
{
"exp": 1749261108,
"hostname": "na.teslemetry.com",
"port": 4431,
"prefer_typed": true,
"pending": false,
"fields": {
"ChargeAmps": { "interval_seconds": 60 }
}
}

22
tests/components/teslemetry/fixtures/metadata.json Normal file
View File

@ -0,0 +1,22 @@
{
"uid": "abc-123",
"region": "NA",
"scopes": [
"openid",
"offline_access",
"user_data",
"vehicle_device_data",
"vehicle_cmds",
"vehicle_charging_cmds",
"energy_device_data",
"energy_cmds"
],
"vehicles": {
"LRW3F7EK4NC700000": {
"access": true,
"polling": true,
"proxy": true,
"firmware": "2024.38.7"
}
}
}

32
tests/components/teslemetry/snapshots/test_sensor.ambr
View File

@ -2414,6 +2414,9 @@
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,
@ -3843,7 +3846,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
'state': 'unavailable',
})
# ---
# name: test_sensors[sensor.test_speed-statealt]
@ -3859,7 +3862,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': '0',
'state': 'unavailable',
})
# ---
# name: test_sensors[sensor.test_state_of_charge_at_arrival-entry]
@ -3910,7 +3913,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
'state': 'unavailable',
})
# ---
# name: test_sensors[sensor.test_state_of_charge_at_arrival-statealt]
@ -3926,7 +3929,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
'state': 'unavailable',
})
# ---
# name: test_sensors[sensor.test_time_to_arrival-entry]
@ -4977,3 +4980,24 @@
'state': 'unknown',
})
# ---
# name: test_sensors_streaming[sensor.test_battery_level-state]
'90'
# ---
# name: test_sensors_streaming[sensor.test_charge_cable-state]
'unknown'
# ---
# name: test_sensors_streaming[sensor.test_charge_energy_added-state]
'10'
# ---
# name: test_sensors_streaming[sensor.test_charger_power-state]
'2'
# ---
# name: test_sensors_streaming[sensor.test_charging-state]
'charging'
# ---
# name: test_sensors_streaming[sensor.test_time_to_arrival-state]
'unknown'
# ---
# name: test_sensors_streaming[sensor.test_time_to_full_charge-state]
'unknown'
# ---

37
tests/components/teslemetry/test_init.py
View File

@ -142,13 +142,13 @@ async def test_energy_history_refresh_error(
async def test_vehicle_stream(
hass: HomeAssistant,
mock_listen: AsyncMock,
mock_add_listener: AsyncMock,
snapshot: SnapshotAssertion,
) -> None:
"""Test vehicle stream events."""
entry = await setup_platform(hass, [Platform.BINARY_SENSOR])
mock_listen.assert_called_once()
await setup_platform(hass, [Platform.BINARY_SENSOR])
mock_add_listener.assert_called()
state = hass.states.get("binary_sensor.test_status")
assert state.state == STATE_ON
@ -156,28 +156,25 @@ async def test_vehicle_stream(
state = hass.states.get("binary_sensor.test_user_present")
assert state.state == STATE_OFF
runtime_data: TeslemetryData = entry.runtime_data
for listener, _ in runtime_data.vehicles[0].stream._listeners.values():
listener(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"vehicle_data": VEHICLE_DATA_ALT["response"],
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"vehicle_data": VEHICLE_DATA_ALT["response"],
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_user_present")
assert state.state == STATE_ON
for listener, _ in runtime_data.vehicles[0].stream._listeners.values():
listener(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"state": "offline",
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"state": "offline",
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
await hass.async_block_till_done()
state = hass.states.get("binary_sensor.test_status")

62
tests/components/teslemetry/test_sensor.py
View File

@ -1,10 +1,11 @@
"""Test the Teslemetry sensor platform."""
from unittest.mock import AsyncMock
from unittest.mock import AsyncMock, patch
from freezegun.api import FrozenDateTimeFactory
import pytest
from syrupy.assertion import SnapshotAssertion
from teslemetry_stream import Signal
from homeassistant.components.teslemetry.coordinator import VEHICLE_INTERVAL
from homeassistant.const import Platform
@ -25,11 +26,15 @@ async def test_sensors(
freezer: FrozenDateTimeFactory,
mock_vehicle_data: AsyncMock,
) -> None:
"""Tests that the sensor entities are correct."""
"""Tests that the sensor entities with the legacy polling are correct."""
freezer.move_to("2024-01-01 00:00:00+00:00")
entry = await setup_platform(hass, [Platform.SENSOR])
# Force the vehicle to use polling
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.pre2021", return_value=True
):
entry = await setup_platform(hass, [Platform.SENSOR])
assert_entities(hass, entry.entry_id, entity_registry, snapshot)
@ -40,3 +45,54 @@ async def test_sensors(
await hass.async_block_till_done()
assert_entities_alt(hass, entry.entry_id, entity_registry, snapshot)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_sensors_streaming(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
freezer: FrozenDateTimeFactory,
mock_vehicle_data: AsyncMock,
mock_add_listener: AsyncMock,
) -> None:
"""Tests that the sensor entities with streaming are correct."""
freezer.move_to("2024-01-01 00:00:00+00:00")
entry = await setup_platform(hass, [Platform.SENSOR])
# Stream update
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"data": {
Signal.DETAILED_CHARGE_STATE: "DetailedChargeStateCharging",
Signal.BATTERY_LEVEL: 90,
Signal.AC_CHARGING_ENERGY_IN: 10,
Signal.AC_CHARGING_POWER: 2,
Signal.CHARGING_CABLE_TYPE: None,
Signal.TIME_TO_FULL_CHARGE: 10,
Signal.MINUTES_TO_ARRIVAL: None,
},
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
await hass.async_block_till_done()
# Reload the entry
await hass.config_entries.async_reload(entry.entry_id)
await hass.async_block_till_done()
# Assert the entities restored their values
for entity_id in (
"sensor.test_charging",
"sensor.test_battery_level",
"sensor.test_charge_energy_added",
"sensor.test_charger_power",
"sensor.test_charge_cable",
"sensor.test_time_to_full_charge",
"sensor.test_time_to_arrival",
):
state = hass.states.get(entity_id)
assert state.state == snapshot(name=f"{entity_id}-state")