jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-04-25 01:38:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add streaming to Teslemetry cover platform (#135660)

Browse Source
This commit is contained in:
Brett Adams 2025-01-18 02:38:03 +10:00 committed by GitHub
parent 9e0df89bee
commit 24bb623567
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 540 additions and 164 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -128,6 +128,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
{"vin": vin},
)
firmware = vehicle_metadata[vin].get("firmware", "Unknown")
stream_vehicle = stream.get_vehicle(vin)
vehicles.append(
TeslemetryVehicleData(
@ -135,6 +136,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: TeslemetryConfigEntry) -
config_entry=entry,
coordinator=coordinator,
stream=stream,
stream_vehicle=stream_vehicle,
vin=vin,
firmware=firmware,
device=device,
@ -285,8 +287,9 @@ async def async_setup_stream(
):
"""Set up the stream for a vehicle."""
vehicle_stream = vehicle.stream.get_vehicle(vehicle.vin)
await vehicle_stream.get_config()
await vehicle.stream_vehicle.get_config()
entry.async_create_background_task(
hass, vehicle_stream.prefer_typed(True), f"Prefer typed for {vehicle.vin}"
hass,
vehicle.stream_vehicle.prefer_typed(True),
f"Prefer typed for {vehicle.vin}",
)

388
homeassistant/components/teslemetry/cover.py
View File

@ -2,9 +2,12 @@
from __future__ import annotations
from itertools import chain
from typing import Any
from tesla_fleet_api.const import Scope, SunRoofCommand, Trunk, WindowCommand
from teslemetry_stream import Signal
from teslemetry_stream.const import WindowState
from homeassistant.components.cover import (
CoverDeviceClass,
@ -13,9 +16,14 @@ from homeassistant.components.cover import (
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.restore_state import RestoreEntity
from . import TeslemetryConfigEntry
from .entity import TeslemetryVehicleEntity
from .entity import (
TeslemetryRootEntity,
TeslemetryVehicleEntity,
TeslemetryVehicleStreamEntity,
)
from .helpers import handle_vehicle_command
from .models import TeslemetryVehicleData
@ -33,30 +41,95 @@ async def async_setup_entry(
"""Set up the Teslemetry cover platform from a config entry."""
async_add_entities(
klass(vehicle, entry.runtime_data.scopes)
for (klass) in (
TeslemetryWindowEntity,
TeslemetryChargePortEntity,
TeslemetryFrontTrunkEntity,
TeslemetryRearTrunkEntity,
TeslemetrySunroofEntity,
chain(
(
TeslemetryPollingWindowEntity(vehicle, entry.runtime_data.scopes)
if vehicle.api.pre2021 or vehicle.firmware < "2024.26"
else TeslemetryStreamingWindowEntity(vehicle, entry.runtime_data.scopes)
for vehicle in entry.runtime_data.vehicles
),
(
TeslemetryPollingChargePortEntity(vehicle, entry.runtime_data.scopes)
if vehicle.api.pre2021 or vehicle.firmware < "2024.44.25"
else TeslemetryStreamingChargePortEntity(
vehicle, entry.runtime_data.scopes
)
for vehicle in entry.runtime_data.vehicles
),
(
TeslemetryPollingFrontTrunkEntity(vehicle, entry.runtime_data.scopes)
if vehicle.api.pre2021 or vehicle.firmware < "2024.26"
else TeslemetryStreamingFrontTrunkEntity(
vehicle, entry.runtime_data.scopes
)
for vehicle in entry.runtime_data.vehicles
),
(
TeslemetryPollingRearTrunkEntity(vehicle, entry.runtime_data.scopes)
if vehicle.api.pre2021 or vehicle.firmware < "2024.26"
else TeslemetryStreamingRearTrunkEntity(
vehicle, entry.runtime_data.scopes
)
for vehicle in entry.runtime_data.vehicles
),
(
TeslemetrySunroofEntity(vehicle, entry.runtime_data.scopes)
for vehicle in entry.runtime_data.vehicles
if vehicle.coordinator.data.get("vehicle_config_sun_roof_installed")
),
)
for vehicle in entry.runtime_data.vehicles
)
class TeslemetryWindowEntity(TeslemetryVehicleEntity, CoverEntity):
"""Cover entity for the windows."""
class CoverRestoreEntity(RestoreEntity, CoverEntity):
"""Restore class for cover entities."""
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
await super().async_added_to_hass()
if (state := await self.async_get_last_state()) is not None:
if state.state == "open":
self._attr_is_closed = False
elif state.state == "closed":
self._attr_is_closed = True
class TeslemetryWindowEntity(TeslemetryRootEntity, CoverEntity):
"""Base class for window cover entities."""
_attr_device_class = CoverDeviceClass.WINDOW
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
async def async_open_cover(self, **kwargs: Any) -> None:
"""Vent windows."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(
self.api.window_control(command=WindowCommand.VENT)
)
self._attr_is_closed = False
self.async_write_ha_state()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close windows."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await handle_vehicle_command(
self.api.window_control(command=WindowCommand.CLOSE)
)
self._attr_is_closed = True
self.async_write_ha_state()
class TeslemetryPollingWindowEntity(
TeslemetryVehicleEntity, TeslemetryWindowEntity, CoverEntity
):
"""Polling cover entity for windows."""
def __init__(self, data: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the cover."""
super().__init__(data, "windows")
self.scoped = Scope.VEHICLE_CMDS in scopes
self._attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
)
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
@ -67,38 +140,108 @@ class TeslemetryWindowEntity(TeslemetryVehicleEntity, CoverEntity):
rd = self.get("vehicle_state_rd_window")
rp = self.get("vehicle_state_rp_window")
# Any open set to open
if OPEN in (fd, fp, rd, rp):
self._attr_is_closed = False
# All closed set to closed
elif CLOSED == fd == fp == rd == rp:
elif None in (fd, fp, rd, rp):
self._attr_is_closed = None
else:
self._attr_is_closed = True
async def async_open_cover(self, **kwargs: Any) -> None:
"""Vent windows."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(
self.api.window_control(command=WindowCommand.VENT)
class TeslemetryStreamingWindowEntity(
TeslemetryVehicleStreamEntity, TeslemetryWindowEntity, CoverRestoreEntity
):
"""Streaming cover entity for windows."""
fd: bool | None = None
fp: bool | None = None
rd: bool | None = None
rp: bool | None = None
def __init__(self, data: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the cover."""
super().__init__(
data,
"windows",
)
self.scoped = Scope.VEHICLE_CMDS in scopes
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
self._attr_is_closed = None
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}},
)
)
for signal in (
Signal.FD_WINDOW,
Signal.FP_WINDOW,
Signal.RD_WINDOW,
Signal.RP_WINDOW,
):
self.vehicle.config_entry.async_create_background_task(
self.hass,
self.add_field(signal),
f"Adding field {signal} to {self.vehicle.vin}",
)
def _handle_stream_update(self, data) -> None:
"""Update the entity attributes."""
if value := data.get(Signal.FD_WINDOW):
self.fd = WindowState.get(value) == "closed"
if value := data.get(Signal.FP_WINDOW):
self.fp = WindowState.get(value) == "closed"
if value := data.get(Signal.RD_WINDOW):
self.rd = WindowState.get(value) == "closed"
if value := data.get(Signal.RP_WINDOW):
self.rp = WindowState.get(value) == "closed"
if False in (self.fd, self.fp, self.rd, self.rp):
self._attr_is_closed = False
elif None in (self.fd, self.fp, self.rd, self.rp):
self._attr_is_closed = None
else:
self._attr_is_closed = True
self.async_write_ha_state()
class TeslemetryChargePortEntity(
TeslemetryRootEntity,
CoverEntity,
):
"""Base class for for charge port cover entities."""
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open charge port."""
self.raise_for_scope(Scope.VEHICLE_CHARGING_CMDS)
await handle_vehicle_command(self.api.charge_port_door_open())
self._attr_is_closed = False
self.async_write_ha_state()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close windows."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(
self.api.window_control(command=WindowCommand.CLOSE)
)
"""Close charge port."""
self.raise_for_scope(Scope.VEHICLE_CHARGING_CMDS)
await handle_vehicle_command(self.api.charge_port_door_close())
self._attr_is_closed = True
self.async_write_ha_state()
class TeslemetryChargePortEntity(TeslemetryVehicleEntity, CoverEntity):
"""Cover entity for the charge port."""
_attr_device_class = CoverDeviceClass.DOOR
class TeslemetryPollingChargePortEntity(
TeslemetryVehicleEntity, TeslemetryChargePortEntity
):
"""Polling cover entity for the charge port."""
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the cover."""
@ -117,75 +260,123 @@ class TeslemetryChargePortEntity(TeslemetryVehicleEntity, CoverEntity):
"""Update the entity attributes."""
self._attr_is_closed = not self._value
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open charge port."""
self.raise_for_scope(Scope.VEHICLE_CHARGING_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.charge_port_door_open())
self._attr_is_closed = False
self.async_write_ha_state()
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close charge port."""
self.raise_for_scope(Scope.VEHICLE_CHARGING_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.charge_port_door_close())
self._attr_is_closed = True
self.async_write_ha_state()
class TeslemetryFrontTrunkEntity(TeslemetryVehicleEntity, CoverEntity):
"""Cover entity for the front trunk."""
_attr_device_class = CoverDeviceClass.DOOR
class TeslemetryStreamingChargePortEntity(
TeslemetryVehicleStreamEntity, TeslemetryChargePortEntity, CoverRestoreEntity
):
"""Streaming cover entity for the charge port."""
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the cover."""
super().__init__(vehicle, "vehicle_state_ft")
self.scoped = Scope.VEHICLE_CMDS in scopes
self._attr_supported_features = CoverEntityFeature.OPEN
"""Initialize the sensor."""
super().__init__(
vehicle,
"charge_state_charge_port_door_open",
)
self.scoped = any(
scope in scopes
for scope in (Scope.VEHICLE_CMDS, Scope.VEHICLE_CHARGING_CMDS)
)
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
self._attr_is_closed = None
def _async_update_attrs(self) -> None:
"""Update the entity attributes."""
self._attr_is_closed = self._value == CLOSED
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.vehicle.stream_vehicle.listen_ChargePortDoorOpen(
self._async_value_from_stream
)
)
self.vehicle.config_entry.async_create_background_task(
self.hass,
self.add_field(Signal.CHARGE_PORT_DOOR_OPEN),
f"Adding field {Signal.CHARGE_PORT_DOOR_OPEN} to {self.vehicle.vin}",
)
def _async_value_from_stream(self, value: bool | None) -> None:
"""Update the value of the entity."""
self._attr_is_closed = None if value is None else not value
self.async_write_ha_state()
class TeslemetryFrontTrunkEntity(TeslemetryRootEntity, CoverEntity):
"""Base class for the front trunk cover entities."""
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = CoverEntityFeature.OPEN
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open front trunk."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.actuate_trunk(Trunk.FRONT))
self._attr_is_closed = False
self.async_write_ha_state()
# In the future this could be extended to add aftermarket close support through a option flow
class TeslemetryRearTrunkEntity(TeslemetryVehicleEntity, CoverEntity):
"""Cover entity for the rear trunk."""
_attr_device_class = CoverDeviceClass.DOOR
class TeslemetryPollingFrontTrunkEntity(
TeslemetryVehicleEntity, TeslemetryFrontTrunkEntity
):
"""Polling cover entity for the front trunk."""
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the cover."""
super().__init__(vehicle, "vehicle_state_rt")
self.scoped = Scope.VEHICLE_CMDS in scopes
self._attr_supported_features = (
CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
)
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
super().__init__(vehicle, "vehicle_state_ft")
def _async_update_attrs(self) -> None:
"""Update the entity attributes."""
self._attr_is_closed = self._value == CLOSED
class TeslemetryStreamingFrontTrunkEntity(
TeslemetryVehicleStreamEntity, TeslemetryFrontTrunkEntity, CoverRestoreEntity
):
"""Streaming cover entity for the front trunk."""
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the sensor."""
super().__init__(vehicle, "vehicle_state_ft")
self.scoped = Scope.VEHICLE_CMDS in scopes
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
self._attr_is_closed = None
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.vehicle.stream_vehicle.listen_TrunkFront(self._async_value_from_stream)
)
self.vehicle.config_entry.async_create_background_task(
self.hass,
self.add_field(Signal.DOOR_STATE),
f"Adding field {Signal.DOOR_STATE} to {self.vehicle.vin}",
)
def _async_value_from_stream(self, value: bool | None) -> None:
"""Update the entity attributes."""
self._attr_is_closed = None if value is None else not value
self.async_write_ha_state()
class TeslemetryRearTrunkEntity(TeslemetryRootEntity, CoverEntity):
"""Cover entity for the rear trunk."""
_attr_device_class = CoverDeviceClass.DOOR
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open rear trunk."""
if self.is_closed is not False:
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.actuate_trunk(Trunk.REAR))
self._attr_is_closed = False
self.async_write_ha_state()
@ -194,12 +385,60 @@ class TeslemetryRearTrunkEntity(TeslemetryVehicleEntity, CoverEntity):
"""Close rear trunk."""
if self.is_closed is not True:
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.actuate_trunk(Trunk.REAR))
self._attr_is_closed = True
self.async_write_ha_state()
class TeslemetryPollingRearTrunkEntity(
TeslemetryVehicleEntity, TeslemetryRearTrunkEntity
):
"""Base class for the rear trunk cover entities."""
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the sensor."""
self.scoped = Scope.VEHICLE_CMDS in scopes
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
super().__init__(vehicle, "vehicle_state_rt")
def _async_update_attrs(self) -> None:
"""Update the entity attributes."""
self._attr_is_closed = self._value == CLOSED
class TeslemetryStreamingRearTrunkEntity(
TeslemetryVehicleStreamEntity, TeslemetryRearTrunkEntity, CoverRestoreEntity
):
"""Polling cover entity for the rear trunk."""
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the cover."""
super().__init__(vehicle, "vehicle_state_rt")
self.scoped = Scope.VEHICLE_CMDS in scopes
if not self.scoped:
self._attr_supported_features = CoverEntityFeature(0)
self._attr_is_closed = None
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.vehicle.stream_vehicle.listen_TrunkRear(self._async_value_from_stream)
)
self.vehicle.config_entry.async_create_background_task(
self.hass,
self.add_field(Signal.DOOR_STATE),
f"Adding field {Signal.DOOR_STATE} to {self.vehicle.vin}",
)
def _async_value_from_stream(self, value: bool | None) -> None:
"""Update the entity attributes."""
self._attr_is_closed = None if value is None else not value
class TeslemetrySunroofEntity(TeslemetryVehicleEntity, CoverEntity):
"""Cover entity for the sunroof."""
@ -210,7 +449,7 @@ class TeslemetrySunroofEntity(TeslemetryVehicleEntity, CoverEntity):
_attr_entity_registry_enabled_default = False
def __init__(self, vehicle: TeslemetryVehicleData, scopes: list[Scope]) -> None:
"""Initialize the sensor."""
"""Initialize the cover."""
super().__init__(vehicle, "vehicle_state_sun_roof_state")
self.scoped = Scope.VEHICLE_CMDS in scopes
@ -232,7 +471,6 @@ class TeslemetrySunroofEntity(TeslemetryVehicleEntity, CoverEntity):
async def async_open_cover(self, **kwargs: Any) -> None:
"""Open sunroof."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.VENT))
self._attr_is_closed = False
self.async_write_ha_state()
@ -240,7 +478,6 @@ class TeslemetrySunroofEntity(TeslemetryVehicleEntity, CoverEntity):
async def async_close_cover(self, **kwargs: Any) -> None:
"""Close sunroof."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.CLOSE))
self._attr_is_closed = True
self.async_write_ha_state()
@ -248,7 +485,6 @@ class TeslemetrySunroofEntity(TeslemetryVehicleEntity, CoverEntity):
async def async_stop_cover(self, **kwargs: Any) -> None:
"""Close sunroof."""
self.raise_for_scope(Scope.VEHICLE_CMDS)
await self.wake_up_if_asleep()
await handle_vehicle_command(self.api.sun_roof_control(SunRoofCommand.STOP))
self._attr_is_closed = False
self.async_write_ha_state()

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

@ -3,6 +3,7 @@
from abc import abstractmethod
from typing import Any
from propcache import cached_property
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import Scope
from teslemetry_stream import Signal
@ -23,18 +24,33 @@ from .helpers import wake_up_vehicle
from .models import TeslemetryEnergyData, TeslemetryVehicleData
class TeslemetryRootEntity(Entity):
"""Parent class for all Teslemetry entities."""
_attr_has_entity_name = True
scoped: bool
api: VehicleSpecific | EnergySpecific
def raise_for_scope(self, scope: Scope):
"""Raise an error if a scope is not available."""
if not self.scoped:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="missing_scope",
translation_placeholders={"scope": scope},
)
class TeslemetryEntity(
TeslemetryRootEntity,
CoordinatorEntity[
TeslemetryVehicleDataCoordinator
| TeslemetryEnergyHistoryCoordinator
| TeslemetryEnergySiteLiveCoordinator
| TeslemetryEnergySiteInfoCoordinator
]
],
):
"""Parent class for all Teslemetry entities."""
_attr_has_entity_name = True
scoped: bool
"""Parent class for all Teslemetry Coordinator entities."""
def __init__(
self,
@ -84,15 +100,6 @@ class TeslemetryEntity(
def _async_update_attrs(self) -> None:
"""Update the attributes of the entity."""
def raise_for_scope(self, scope: Scope):
"""Raise an error if a scope is not available."""
if not self.scoped:
raise ServiceValidationError(
translation_domain=DOMAIN,
translation_key="missing_scope",
translation_placeholders={"scope": scope},
)
class TeslemetryVehicleEntity(TeslemetryEntity):
"""Parent class for Teslemetry Vehicle entities."""
@ -239,13 +246,11 @@ class TeslemetryWallConnectorEntity(TeslemetryEntity):
)
class TeslemetryVehicleStreamEntity(Entity):
class TeslemetryVehicleStreamEntity(TeslemetryRootEntity):
"""Parent class for Teslemetry Vehicle Stream entities."""
_attr_has_entity_name = True
def __init__(
self, data: TeslemetryVehicleData, key: str, streaming_key: Signal
self, data: TeslemetryVehicleData, key: str, streaming_key: Signal | None = None
) -> None:
"""Initialize common aspects of a Teslemetry entity."""
self.streaming_key = streaming_key
@ -263,17 +268,18 @@ class TeslemetryVehicleStreamEntity(Entity):
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}},
if self.streaming_key:
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}",
)
)
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."""
@ -283,3 +289,8 @@ class TeslemetryVehicleStreamEntity(Entity):
def _async_value_from_stream(self, value: Any) -> None:
"""Update the entity with the latest value from the stream."""
raise NotImplementedError
@cached_property
def available(self) -> bool:
"""Return True if entity is available."""
return self.stream.connected

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

@ -8,7 +8,7 @@ from dataclasses import dataclass
from tesla_fleet_api import EnergySpecific, VehicleSpecific
from tesla_fleet_api.const import Scope
from teslemetry_stream import TeslemetryStream
from teslemetry_stream import TeslemetryStream, TeslemetryStreamVehicle
from homeassistant.config_entries import ConfigEntry
from homeassistant.helpers.device_registry import DeviceInfo
@ -38,6 +38,7 @@ class TeslemetryVehicleData:
config_entry: ConfigEntry
coordinator: TeslemetryVehicleDataCoordinator
stream: TeslemetryStream
stream_vehicle: TeslemetryStreamVehicle
vin: str
firmware: str
device: DeviceInfo

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

@ -49,6 +49,15 @@ def mock_vehicle_data() -> Generator[AsyncMock]:
yield mock_vehicle_data
@pytest.fixture
def mock_legacy():
"""Mock Tesla Fleet Api products method."""
with patch(
"homeassistant.components.teslemetry.VehicleSpecific.pre2021", return_value=True
) as mock_pre2021:
yield mock_pre2021
@pytest.fixture(autouse=True)
def mock_wake_up():
"""Mock Tesla Fleet API Vehicle Specific wake_up method."""

2
tests/components/teslemetry/fixtures/metadata.json
View File

@ -16,7 +16,7 @@
"access": true,
"polling": true,
"proxy": true,
"firmware": "2024.38.7"
"firmware": "2024.44.25"
}
}
}

2
tests/components/teslemetry/fixtures/vehicle_data.json
View File

@ -192,7 +192,7 @@
"api_version": 71,
"autopark_state_v2": "unavailable",
"calendar_supported": true,
"car_version": "2023.44.30.8 06f534d46010",
"car_version": "2024.44.25 06f534d46010",
"center_display_state": 0,
"dashcam_clip_save_available": true,
"dashcam_state": "Recording",

2
tests/components/teslemetry/fixtures/vehicle_data_alt.json
View File

@ -190,7 +190,7 @@
"api_version": 71,
"autopark_state_v2": "unavailable",
"calendar_supported": true,
"car_version": "2023.44.30.8 06f534d46010",
"car_version": "2024.44.25 06f534d46010",
"center_display_state": 0,
"dashcam_clip_save_available": true,
"dashcam_state": "Recording",

84
tests/components/teslemetry/snapshots/test_cover.ambr
View File

@ -335,54 +335,6 @@
'state': 'open',
})
# ---
# name: test_cover_alt[cover.test_sunroof-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'cover',
'entity_category': None,
'entity_id': 'cover.test_sunroof',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': <CoverDeviceClass.WINDOW: 'window'>,
'original_icon': None,
'original_name': 'Sunroof',
'platform': 'teslemetry',
'previous_unique_id': None,
'supported_features': <CoverEntityFeature: 11>,
'translation_key': 'vehicle_state_sun_roof_state',
'unique_id': 'LRW3F7EK4NC700000-vehicle_state_sun_roof_state',
'unit_of_measurement': None,
})
# ---
# name: test_cover_alt[cover.test_sunroof-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'device_class': 'window',
'friendly_name': 'Test Sunroof',
'supported_features': <CoverEntityFeature: 11>,
}),
'context': <ANY>,
'entity_id': 'cover.test_sunroof',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'unknown',
})
# ---
# name: test_cover_alt[cover.test_trunk-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
@ -719,3 +671,39 @@
'state': 'closed',
})
# ---
# name: test_cover_streaming[cover.test_charge_port_door-closed]
'closed'
# ---
# name: test_cover_streaming[cover.test_charge_port_door-open]
'closed'
# ---
# name: test_cover_streaming[cover.test_charge_port_door-unknown]
'unknown'
# ---
# name: test_cover_streaming[cover.test_frunk-closed]
'unknown'
# ---
# name: test_cover_streaming[cover.test_frunk-open]
'unknown'
# ---
# name: test_cover_streaming[cover.test_frunk-unknown]
'unknown'
# ---
# name: test_cover_streaming[cover.test_trunk-closed]
'unknown'
# ---
# name: test_cover_streaming[cover.test_trunk-open]
'unknown'
# ---
# name: test_cover_streaming[cover.test_trunk-unknown]
'unknown'
# ---
# name: test_cover_streaming[cover.test_windows-closed]
'unknown'
# ---
# name: test_cover_streaming[cover.test_windows-open]
'unknown'
# ---
# name: test_cover_streaming[cover.test_windows-unknown]
'unknown'
# ---

2
tests/components/teslemetry/snapshots/test_diagnostics.ambr
View File

@ -352,7 +352,7 @@
'vehicle_state_api_version': 71,
'vehicle_state_autopark_state_v2': 'unavailable',
'vehicle_state_calendar_supported': True,
'vehicle_state_car_version': '2023.44.30.8 06f534d46010',
'vehicle_state_car_version': '2024.44.25 06f534d46010',
'vehicle_state_center_display_state': 0,
'vehicle_state_dashcam_clip_save_available': True,
'vehicle_state_dashcam_state': 'Recording',

8
tests/components/teslemetry/snapshots/test_update.ambr
View File

@ -40,7 +40,7 @@
'entity_picture': 'https://brands.home-assistant.io/_/teslemetry/icon.png',
'friendly_name': 'Test Update',
'in_progress': False,
'installed_version': '2023.44.30.8',
'installed_version': '2024.44.25',
'latest_version': '2024.12.0.0',
'release_summary': None,
'release_url': None,
@ -54,7 +54,7 @@
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
'state': 'off',
})
# ---
# name: test_update_alt[update.test_update-entry]
@ -98,8 +98,8 @@
'entity_picture': 'https://brands.home-assistant.io/_/teslemetry/icon.png',
'friendly_name': 'Test Update',
'in_progress': False,
'installed_version': '2023.44.30.8',
'latest_version': '2023.44.30.8',
'installed_version': '2024.44.25',
'latest_version': '2024.44.25',
'release_summary': None,
'release_url': None,
'skipped_version': None,

128
tests/components/teslemetry/test_cover.py
View File

@ -4,6 +4,7 @@ from unittest.mock import AsyncMock, patch
import pytest
from syrupy.assertion import SnapshotAssertion
from teslemetry_stream import Signal
from homeassistant.components.cover import (
DOMAIN as COVER_DOMAIN,
@ -25,6 +26,7 @@ async def test_cover(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the cover entities are correct."""
@ -38,6 +40,7 @@ async def test_cover_alt(
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data: AsyncMock,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the cover entities are correct with alternate values."""
@ -52,6 +55,7 @@ async def test_cover_noscope(
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_metadata: AsyncMock,
mock_legacy: AsyncMock,
) -> None:
"""Tests that the cover entities are correct without scopes."""
@ -215,3 +219,127 @@ async def test_cover_services(
state = hass.states.get(entity_id)
assert state
assert state.state == CoverState.CLOSED
async def test_cover_streaming(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_vehicle_data: AsyncMock,
mock_add_listener: AsyncMock,
) -> None:
"""Tests that the binary sensor entities with streaming are correct."""
entry = await setup_platform(hass, [Platform.COVER])
# Stream update
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"data": {
Signal.FD_WINDOW: "WindowStateClosed",
Signal.FP_WINDOW: "WindowStateClosed",
Signal.RD_WINDOW: "WindowStateClosed",
Signal.RP_WINDOW: "WindowStateClosed",
Signal.CHARGE_PORT_DOOR_OPEN: False,
Signal.DOOR_STATE: {
"DoorState": {
"DriverFront": False,
"DriverRear": False,
"PassengerFront": False,
"PassengerRear": False,
"TrunkFront": False,
"TrunkRear": False,
}
},
},
"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 (
"cover.test_windows",
"cover.test_charge_port_door",
"cover.test_frunk",
"cover.test_trunk",
):
state = hass.states.get(entity_id)
assert state.state == snapshot(name=f"{entity_id}-closed")
# Send some alternative data with everything open
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"data": {
Signal.FD_WINDOW: "WindowStateOpened",
Signal.FP_WINDOW: "WindowStateOpened",
Signal.RD_WINDOW: "WindowStateOpened",
Signal.RP_WINDOW: "WindowStateOpened",
Signal.CHARGE_PORT_DOOR_OPEN: False,
Signal.DOOR_STATE: {
"DoorState": {
"DriverFront": True,
"DriverRear": True,
"PassengerFront": True,
"PassengerRear": True,
"TrunkFront": True,
"TrunkRear": True,
}
},
},
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
await hass.async_block_till_done()
# Assert the entities get new values
for entity_id in (
"cover.test_windows",
"cover.test_charge_port_door",
"cover.test_frunk",
"cover.test_trunk",
):
state = hass.states.get(entity_id)
assert state.state == snapshot(name=f"{entity_id}-open")
# Send some alternative data with everything unknown
mock_add_listener.send(
{
"vin": VEHICLE_DATA_ALT["response"]["vin"],
"data": {
Signal.FD_WINDOW: "WindowStateUnknown",
Signal.FP_WINDOW: "WindowStateUnknown",
Signal.RD_WINDOW: "WindowStateUnknown",
Signal.RP_WINDOW: "WindowStateUnknown",
Signal.CHARGE_PORT_DOOR_OPEN: None,
Signal.DOOR_STATE: {
"DoorState": {
"DriverFront": None,
"DriverRear": None,
"PassengerFront": None,
"PassengerRear": None,
"TrunkFront": None,
"TrunkRear": None,
}
},
},
"createdAt": "2024-10-04T10:45:17.537Z",
}
)
await hass.async_block_till_done()
# Assert the entities get UNKNOWN values
for entity_id in (
"cover.test_windows",
"cover.test_charge_port_door",
"cover.test_frunk",
"cover.test_trunk",
):
state = hass.states.get(entity_id)
assert state.state == snapshot(name=f"{entity_id}-unknown")