Add typed listeners to Teslemetry binary sensor platform (#142238)

This commit is contained in:
Brett Adams 2025-04-19 19:29:14 +10:00 committed by GitHub
parent 44450f9d7d
commit 7c3df46570
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 86 additions and 60 deletions

View File

@ -6,8 +6,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from typing import cast
from teslemetry_stream import Signal
from teslemetry_stream.const import WindowState
from teslemetry_stream.vehicle import TeslemetryStreamVehicle
from homeassistant.components.binary_sensor import (
BinarySensorDeviceClass,
@ -32,6 +31,12 @@ from .models import TeslemetryEnergyData, TeslemetryVehicleData
PARALLEL_UPDATES = 0
WINDOW_STATES = {
"Opened": True,
"PartiallyOpen": True,
"Closed": False,
}
@dataclass(frozen=True, kw_only=True)
class TeslemetryBinarySensorEntityDescription(BinarySensorEntityDescription):
@ -39,11 +44,14 @@ class TeslemetryBinarySensorEntityDescription(BinarySensorEntityDescription):
polling_value_fn: Callable[[StateType], bool | None] = bool
polling: bool = False
streaming_key: Signal | None = None
streaming_listener: (
Callable[
[TeslemetryStreamVehicle, Callable[[bool | None], None]],
Callable[[], None],
]
| None
) = None
streaming_firmware: str = "2024.26"
streaming_value_fn: Callable[[StateType], bool | None] = (
lambda x: x is True or x == "true"
)
VEHICLE_DESCRIPTIONS: tuple[TeslemetryBinarySensorEntityDescription, ...] = (
@ -56,7 +64,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryBinarySensorEntityDescription, ...] = (
TeslemetryBinarySensorEntityDescription(
key="charge_state_battery_heater_on",
polling=True,
streaming_key=Signal.BATTERY_HEATER_ON,
streaming_listener=lambda x, y: x.listen_BatteryHeaterOn(y),
device_class=BinarySensorDeviceClass.HEAT,
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
@ -64,15 +72,16 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryBinarySensorEntityDescription, ...] = (
TeslemetryBinarySensorEntityDescription(
key="charge_state_charger_phases",
polling=True,
streaming_key=Signal.CHARGER_PHASES,
streaming_listener=lambda x, y: x.listen_ChargerPhases(
lambda z: y(None if z is None else z > 1)
),
polling_value_fn=lambda x: cast(int, x) > 1,
streaming_value_fn=lambda x: cast(int, x) > 1,
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="charge_state_preconditioning_enabled",
polling=True,
streaming_key=Signal.PRECONDITIONING_ENABLED,
streaming_listener=lambda x, y: x.listen_PreconditioningEnabled(y),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
@ -85,7 +94,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryBinarySensorEntityDescription, ...] = (
TeslemetryBinarySensorEntityDescription(
key="charge_state_scheduled_charging_pending",
polling=True,
streaming_key=Signal.SCHEDULED_CHARGING_PENDING,
streaming_listener=lambda x, y: x.listen_ScheduledChargingPending(y),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
@ -153,32 +162,36 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryBinarySensorEntityDescription, ...] = (
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_fd_window",
polling=True,
streaming_key=Signal.FD_WINDOW,
streaming_value_fn=lambda x: WindowState.get(x) != "Closed",
streaming_listener=lambda x, y: x.listen_FrontDriverWindow(
lambda z: y(WINDOW_STATES.get(z))
),
device_class=BinarySensorDeviceClass.WINDOW,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_fp_window",
polling=True,
streaming_key=Signal.FP_WINDOW,
streaming_value_fn=lambda x: WindowState.get(x) != "Closed",
streaming_listener=lambda x, y: x.listen_FrontPassengerWindow(
lambda z: y(WINDOW_STATES.get(z))
),
device_class=BinarySensorDeviceClass.WINDOW,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_rd_window",
polling=True,
streaming_key=Signal.RD_WINDOW,
streaming_value_fn=lambda x: WindowState.get(x) != "Closed",
streaming_listener=lambda x, y: x.listen_RearDriverWindow(
lambda z: y(WINDOW_STATES.get(z))
),
device_class=BinarySensorDeviceClass.WINDOW,
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_rp_window",
polling=True,
streaming_key=Signal.RP_WINDOW,
streaming_value_fn=lambda x: WindowState.get(x) != "Closed",
streaming_listener=lambda x, y: x.listen_RearPassengerWindow(
lambda z: y(WINDOW_STATES.get(z))
),
device_class=BinarySensorDeviceClass.WINDOW,
entity_category=EntityCategory.DIAGNOSTIC,
),
@ -186,180 +199,177 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryBinarySensorEntityDescription, ...] = (
key="vehicle_state_df",
polling=True,
device_class=BinarySensorDeviceClass.DOOR,
streaming_key=Signal.DOOR_STATE,
streaming_value_fn=lambda x: cast(dict, x).get("DriverFront"),
streaming_listener=lambda x, y: x.listen_FrontDriverDoor(y),
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_dr",
polling=True,
device_class=BinarySensorDeviceClass.DOOR,
streaming_key=Signal.DOOR_STATE,
streaming_value_fn=lambda x: cast(dict, x).get("DriverRear"),
streaming_listener=lambda x, y: x.listen_RearDriverDoor(y),
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_pf",
polling=True,
device_class=BinarySensorDeviceClass.DOOR,
streaming_key=Signal.DOOR_STATE,
streaming_value_fn=lambda x: cast(dict, x).get("PassengerFront"),
streaming_listener=lambda x, y: x.listen_FrontPassengerDoor(y),
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="vehicle_state_pr",
polling=True,
device_class=BinarySensorDeviceClass.DOOR,
streaming_key=Signal.DOOR_STATE,
streaming_value_fn=lambda x: cast(dict, x).get("PassengerRear"),
streaming_listener=lambda x, y: x.listen_RearPassengerDoor(y),
entity_category=EntityCategory.DIAGNOSTIC,
),
TeslemetryBinarySensorEntityDescription(
key="automatic_blind_spot_camera",
streaming_key=Signal.AUTOMATIC_BLIND_SPOT_CAMERA,
streaming_listener=lambda x, y: x.listen_AutomaticBlindSpotCamera(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="automatic_emergency_braking_off",
streaming_key=Signal.AUTOMATIC_EMERGENCY_BRAKING_OFF,
streaming_listener=lambda x, y: x.listen_AutomaticEmergencyBrakingOff(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="blind_spot_collision_warning_chime",
streaming_key=Signal.BLIND_SPOT_COLLISION_WARNING_CHIME,
streaming_listener=lambda x, y: x.listen_BlindSpotCollisionWarningChime(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="bms_full_charge_complete",
streaming_key=Signal.BMS_FULL_CHARGE_COMPLETE,
streaming_listener=lambda x, y: x.listen_BmsFullchargecomplete(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="brake_pedal",
streaming_key=Signal.BRAKE_PEDAL,
streaming_listener=lambda x, y: x.listen_BrakePedal(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="charge_port_cold_weather_mode",
streaming_key=Signal.CHARGE_PORT_COLD_WEATHER_MODE,
streaming_listener=lambda x, y: x.listen_ChargePortColdWeatherMode(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="service_mode",
streaming_key=Signal.SERVICE_MODE,
streaming_listener=lambda x, y: x.listen_ServiceMode(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="pin_to_drive_enabled",
streaming_key=Signal.PIN_TO_DRIVE_ENABLED,
streaming_listener=lambda x, y: x.listen_PinToDriveEnabled(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="drive_rail",
streaming_key=Signal.DRIVE_RAIL,
streaming_listener=lambda x, y: x.listen_DriveRail(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="driver_seat_belt",
streaming_key=Signal.DRIVER_SEAT_BELT,
streaming_listener=lambda x, y: x.listen_DriverSeatBelt(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="driver_seat_occupied",
streaming_key=Signal.DRIVER_SEAT_OCCUPIED,
streaming_listener=lambda x, y: x.listen_DriverSeatOccupied(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="passenger_seat_belt",
streaming_key=Signal.PASSENGER_SEAT_BELT,
streaming_listener=lambda x, y: x.listen_PassengerSeatBelt(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="fast_charger_present",
streaming_key=Signal.FAST_CHARGER_PRESENT,
streaming_listener=lambda x, y: x.listen_FastChargerPresent(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="gps_state",
streaming_key=Signal.GPS_STATE,
streaming_listener=lambda x, y: x.listen_GpsState(y),
entity_registry_enabled_default=False,
entity_category=EntityCategory.DIAGNOSTIC,
device_class=BinarySensorDeviceClass.CONNECTIVITY,
),
TeslemetryBinarySensorEntityDescription(
key="guest_mode_enabled",
streaming_key=Signal.GUEST_MODE_ENABLED,
streaming_listener=lambda x, y: x.listen_GuestModeEnabled(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="dc_dc_enable",
streaming_key=Signal.DCDC_ENABLE,
streaming_listener=lambda x, y: x.listen_DCDCEnable(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="emergency_lane_departure_avoidance",
streaming_key=Signal.EMERGENCY_LANE_DEPARTURE_AVOIDANCE,
streaming_listener=lambda x, y: x.listen_EmergencyLaneDepartureAvoidance(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="supercharger_session_trip_planner",
streaming_key=Signal.SUPERCHARGER_SESSION_TRIP_PLANNER,
streaming_listener=lambda x, y: x.listen_SuperchargerSessionTripPlanner(y),
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="wiper_heat_enabled",
streaming_key=Signal.WIPER_HEAT_ENABLED,
streaming_listener=lambda x, y: x.listen_WiperHeatEnabled(y),
streaming_firmware="2024.44.25",
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="rear_display_hvac_enabled",
streaming_key=Signal.REAR_DISPLAY_HVAC_ENABLED,
streaming_listener=lambda x, y: x.listen_RearDisplayHvacEnabled(y),
streaming_firmware="2024.44.25",
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="offroad_lightbar_present",
streaming_key=Signal.OFFROAD_LIGHTBAR_PRESENT,
streaming_listener=lambda x, y: x.listen_OffroadLightbarPresent(y),
streaming_firmware="2024.44.25",
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="homelink_nearby",
streaming_key=Signal.HOMELINK_NEARBY,
streaming_listener=lambda x, y: x.listen_HomelinkNearby(y),
streaming_firmware="2024.44.25",
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="europe_vehicle",
streaming_key=Signal.EUROPE_VEHICLE,
streaming_listener=lambda x, y: x.listen_EuropeVehicle(y),
streaming_firmware="2024.44.25",
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="right_hand_drive",
streaming_key=Signal.RIGHT_HAND_DRIVE,
streaming_listener=lambda x, y: x.listen_RightHandDrive(y),
streaming_firmware="2024.44.25",
entity_registry_enabled_default=False,
),
TeslemetryBinarySensorEntityDescription(
key="located_at_home",
streaming_key=Signal.LOCATED_AT_HOME,
streaming_listener=lambda x, y: x.listen_LocatedAtHome(y),
streaming_firmware="2024.44.32",
),
TeslemetryBinarySensorEntityDescription(
key="located_at_work",
streaming_key=Signal.LOCATED_AT_WORK,
streaming_listener=lambda x, y: x.listen_LocatedAtWork(y),
streaming_firmware="2024.44.32",
),
TeslemetryBinarySensorEntityDescription(
key="located_at_favorite",
streaming_key=Signal.LOCATED_AT_FAVORITE,
streaming_listener=lambda x, y: x.listen_LocatedAtFavorite(y),
streaming_firmware="2024.44.32",
entity_registry_enabled_default=False,
),
)
ENERGY_LIVE_DESCRIPTIONS: tuple[BinarySensorEntityDescription, ...] = (
BinarySensorEntityDescription(key="backup_capable"),
BinarySensorEntityDescription(key="grid_services_active"),
@ -386,7 +396,7 @@ async def async_setup_entry(
for description in VEHICLE_DESCRIPTIONS:
if (
not vehicle.api.pre2021
and description.streaming_key
and description.streaming_listener
and vehicle.firmware >= description.streaming_firmware
):
entities.append(
@ -453,8 +463,7 @@ class TeslemetryVehicleStreamingBinarySensorEntity(
) -> None:
"""Initialize the sensor."""
self.entity_description = description
assert description.streaming_key
super().__init__(data, description.key, description.streaming_key)
super().__init__(data, description.key)
async def async_added_to_hass(self) -> None:
"""Handle entity which will be added."""
@ -462,11 +471,18 @@ class TeslemetryVehicleStreamingBinarySensorEntity(
if (state := await self.async_get_last_state()) is not None:
self._attr_is_on = state.state == STATE_ON
def _async_value_from_stream(self, value) -> None:
assert self.entity_description.streaming_listener
self.async_on_remove(
self.entity_description.streaming_listener(
self.vehicle.stream_vehicle, self._async_value_from_stream
)
)
def _async_value_from_stream(self, value: bool | None) -> None:
"""Update the value of the entity."""
self._attr_available = value is not None
if self._attr_available:
self._attr_is_on = self.entity_description.streaming_value_fn(value)
self._attr_is_on = value
self.async_write_ha_state()
class TeslemetryEnergyLiveBinarySensorEntity(

View File

@ -3290,5 +3290,11 @@
'off'
# ---
# name: test_binary_sensors_streaming[binary_sensor.test_front_passenger_window-state]
'off'
# ---
# name: test_binary_sensors_streaming[binary_sensor.test_rear_driver_window-state]
'off'
# ---
# name: test_binary_sensors_streaming[binary_sensor.test_rear_passenger_window-state]
'on'
# ---

View File

@ -73,6 +73,8 @@ async def test_binary_sensors_streaming(
"data": {
Signal.FD_WINDOW: "WindowStateOpened",
Signal.FP_WINDOW: "INVALID_VALUE",
Signal.RD_WINDOW: "WindowStateClosed",
Signal.RP_WINDOW: "WindowStatePartiallyOpen",
Signal.DOOR_STATE: {
"DoorState": {
"DriverFront": True,
@ -98,6 +100,8 @@ async def test_binary_sensors_streaming(
for entity_id in (
"binary_sensor.test_front_driver_window",
"binary_sensor.test_front_passenger_window",
"binary_sensor.test_rear_driver_window",
"binary_sensor.test_rear_passenger_window",
"binary_sensor.test_front_driver_door",
"binary_sensor.test_front_passenger_door",
"binary_sensor.test_driver_seat_belt",