jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 20:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add typed listeners to Teslemetry sensor platform (#142236)

Browse Source
This commit is contained in:
Brett Adams 2025-04-19 18:56:29 +10:00 committed by GitHub
parent aef266b940
commit c34e280fc2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 57 additions and 42 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,8 +7,7 @@ from dataclasses import dataclass
from datetime import datetime, timedelta
from propcache.api import cached_property
from teslemetry_stream import Signal, TeslemetryStreamVehicle
from teslemetry_stream.const import ShiftState
from teslemetry_stream import TeslemetryStreamVehicle
from homeassistant.components.sensor import (
RestoreSensor,
@ -70,8 +69,13 @@ class TeslemetryVehicleSensorEntityDescription(SensorEntityDescription):
polling: bool = False
polling_value_fn: Callable[[StateType], StateType] = lambda x: x
nullable: bool = False
streaming_key: Signal | None = None
streaming_value_fn: Callable[[str | int | float], StateType] = lambda x: x
streaming_listener: (
Callable[
[TeslemetryStreamVehicle, Callable[[StateType], None]],
Callable[[], None],
]
| None
) = None
streaming_firmware: str = "2024.26"
@ -79,18 +83,17 @@ 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", "")
streaming_listener=lambda x, y: x.listen_DetailedChargeState(
lambda z: None if z is None else y(z.lower())
),
polling_value_fn=lambda value: CHARGE_STATES.get(str(value)),
options=list(CHARGE_STATES.values()),
device_class=SensorDeviceClass.ENUM,
),
TeslemetryVehicleSensorEntityDescription(
key="charge_state_battery_level",
polling=True,
streaming_key=Signal.BATTERY_LEVEL,
streaming_listener=lambda x, y: x.listen_BatteryLevel(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
@ -99,15 +102,17 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_usable_battery_level",
polling=True,
streaming_listener=lambda x, y: x.listen_Soc(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
entity_registry_enabled_default=False,
suggested_display_precision=1,
),
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charge_energy_added",
polling=True,
streaming_key=Signal.AC_CHARGING_ENERGY_IN,
streaming_listener=lambda x, y: x.listen_ACChargingEnergyIn(y),
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfEnergy.KILO_WATT_HOUR,
device_class=SensorDeviceClass.ENERGY,
@ -116,7 +121,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charger_power",
polling=True,
streaming_key=Signal.AC_CHARGING_POWER,
streaming_listener=lambda x, y: x.listen_ACChargingPower(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPower.KILO_WATT,
device_class=SensorDeviceClass.POWER,
@ -124,7 +129,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charger_voltage",
polling=True,
streaming_key=Signal.CHARGER_VOLTAGE,
streaming_listener=lambda x, y: x.listen_ChargerVoltage(y),
streaming_firmware="2024.44.32",
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
@ -134,7 +139,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_charger_actual_current",
polling=True,
streaming_key=Signal.CHARGE_AMPS,
streaming_listener=lambda x, y: x.listen_ChargeAmps(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfElectricCurrent.AMPERE,
device_class=SensorDeviceClass.CURRENT,
@ -151,14 +156,14 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_conn_charge_cable",
polling=True,
streaming_key=Signal.CHARGING_CABLE_TYPE,
streaming_listener=lambda x, y: x.listen_ChargingCableType(y),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
TeslemetryVehicleSensorEntityDescription(
key="charge_state_fast_charger_type",
polling=True,
streaming_key=Signal.FAST_CHARGER_TYPE,
streaming_listener=lambda x, y: x.listen_FastChargerType(y),
entity_category=EntityCategory.DIAGNOSTIC,
entity_registry_enabled_default=False,
),
@ -173,7 +178,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_est_battery_range",
polling=True,
streaming_key=Signal.EST_BATTERY_RANGE,
streaming_listener=lambda x, y: x.listen_EstBatteryRange(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
@ -183,7 +188,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="charge_state_ideal_battery_range",
polling=True,
streaming_key=Signal.IDEAL_BATTERY_RANGE,
streaming_listener=lambda x, y: x.listen_IdealBatteryRange(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
@ -194,7 +199,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
key="drive_state_speed",
polling=True,
polling_value_fn=lambda value: value or 0,
streaming_key=Signal.VEHICLE_SPEED,
streaming_listener=lambda x, y: x.listen_VehicleSpeed(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfSpeed.MILES_PER_HOUR,
device_class=SensorDeviceClass.SPEED,
@ -213,10 +218,11 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="drive_state_shift_state",
polling=True,
nullable=True,
polling_value_fn=lambda x: SHIFT_STATES.get(str(x), "p"),
streaming_key=Signal.GEAR,
streaming_value_fn=lambda x: str(ShiftState.get(x, "P")).lower(),
nullable=True,
streaming_listener=lambda x, y: x.listen_Gear(
lambda z: y("p" if z is None else z.lower())
),
options=list(SHIFT_STATES.values()),
device_class=SensorDeviceClass.ENUM,
entity_registry_enabled_default=False,
@ -224,7 +230,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_odometer",
polling=True,
streaming_key=Signal.ODOMETER,
streaming_listener=lambda x, y: x.listen_Odometer(y),
state_class=SensorStateClass.TOTAL_INCREASING,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
@ -235,7 +241,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_fl",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_FL,
streaming_listener=lambda x, y: x.listen_TpmsPressureFl(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -247,7 +253,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_fr",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_FR,
streaming_listener=lambda x, y: x.listen_TpmsPressureFr(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -259,7 +265,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_rl",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_RL,
streaming_listener=lambda x, y: x.listen_TpmsPressureRl(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -271,7 +277,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="vehicle_state_tpms_pressure_rr",
polling=True,
streaming_key=Signal.TPMS_PRESSURE_RR,
streaming_listener=lambda x, y: x.listen_TpmsPressureRr(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfPressure.BAR,
suggested_unit_of_measurement=UnitOfPressure.PSI,
@ -283,7 +289,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="climate_state_inside_temp",
polling=True,
streaming_key=Signal.INSIDE_TEMP,
streaming_listener=lambda x, y: x.listen_InsideTemp(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
@ -292,7 +298,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="climate_state_outside_temp",
polling=True,
streaming_key=Signal.OUTSIDE_TEMP,
streaming_listener=lambda x, y: x.listen_OutsideTemp(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
device_class=SensorDeviceClass.TEMPERATURE,
@ -321,7 +327,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="drive_state_active_route_traffic_minutes_delay",
polling=True,
streaming_key=Signal.ROUTE_TRAFFIC_MINUTES_DELAY,
streaming_listener=lambda x, y: x.listen_RouteTrafficMinutesDelay(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfTime.MINUTES,
device_class=SensorDeviceClass.DURATION,
@ -330,7 +336,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="drive_state_active_route_energy_at_arrival",
polling=True,
streaming_key=Signal.EXPECTED_ENERGY_PERCENT_AT_TRIP_ARRIVAL,
streaming_listener=lambda x, y: x.listen_ExpectedEnergyPercentAtTripArrival(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=PERCENTAGE,
device_class=SensorDeviceClass.BATTERY,
@ -340,7 +346,7 @@ VEHICLE_DESCRIPTIONS: tuple[TeslemetryVehicleSensorEntityDescription, ...] = (
TeslemetryVehicleSensorEntityDescription(
key="drive_state_active_route_miles_to_arrival",
polling=True,
streaming_key=Signal.MILES_TO_ARRIVAL,
streaming_listener=lambda x, y: x.listen_MilesToArrival(y),
state_class=SensorStateClass.MEASUREMENT,
native_unit_of_measurement=UnitOfLength.MILES,
device_class=SensorDeviceClass.DISTANCE,
@ -358,14 +364,14 @@ class TeslemetryTimeEntityDescription(SensorEntityDescription):
Callable[[], None],
]
streaming_firmware: str = "2024.26"
streaming_value_fn: Callable[[float], float] = lambda x: x
streaming_unit: str
VEHICLE_TIME_DESCRIPTIONS: tuple[TeslemetryTimeEntityDescription, ...] = (
TeslemetryTimeEntityDescription(
key="charge_state_minutes_to_full_charge",
streaming_value_fn=lambda x: x * 60,
streaming_listener=lambda x, y: x.listen_TimeToFullCharge(y),
streaming_unit="hours",
device_class=SensorDeviceClass.TIMESTAMP,
entity_category=EntityCategory.DIAGNOSTIC,
variance=4,
@ -373,6 +379,7 @@ VEHICLE_TIME_DESCRIPTIONS: tuple[TeslemetryTimeEntityDescription, ...] = (
TeslemetryTimeEntityDescription(
key="drive_state_active_route_minutes_to_arrival",
streaming_listener=lambda x, y: x.listen_MinutesToArrival(y),
streaming_unit="minutes",
device_class=SensorDeviceClass.TIMESTAMP,
variance=1,
),
@ -547,7 +554,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(TeslemetryStreamSensorEntity(vehicle, description))
@ -613,8 +620,7 @@ class TeslemetryStreamSensorEntity(TeslemetryVehicleStreamEntity, RestoreSensor)
) -> 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."""
@ -623,17 +629,22 @@ class TeslemetryStreamSensorEntity(TeslemetryVehicleStreamEntity, RestoreSensor)
if (sensor_data := await self.async_get_last_sensor_data()) is not None:
self._attr_native_value = sensor_data.native_value
if self.entity_description.streaming_listener is not None:
self.async_on_remove(
self.entity_description.streaming_listener(
self.vehicle.stream_vehicle, self._async_value_from_stream
)
)
@cached_property
def available(self) -> bool:
"""Return True if entity is available."""
return self.stream.connected
def _async_value_from_stream(self, value) -> None:
def _async_value_from_stream(self, value: StateType) -> None:
"""Update the value of the entity."""
if self.entity_description.nullable or value is not None:
self._attr_native_value = self.entity_description.streaming_value_fn(value)
else:
self._attr_native_value = None
self._attr_native_value = value
self.async_write_ha_state()
class TeslemetryVehicleSensorEntity(TeslemetryVehicleEntity, SensorEntity):
@ -676,7 +687,7 @@ class TeslemetryStreamTimeSensorEntity(TeslemetryVehicleStreamEntity, SensorEnti
self.entity_description = description
self._get_timestamp = ignore_variance(
func=lambda value: dt_util.now()
+ timedelta(minutes=description.streaming_value_fn(value)),
+ timedelta(**{self.entity_description.streaming_unit: value}),
ignored_variance=timedelta(minutes=description.variance),
)
super().__init__(data, description.key)
@ -696,6 +707,7 @@ class TeslemetryStreamTimeSensorEntity(TeslemetryVehicleStreamEntity, SensorEnti
self._attr_native_value = None
else:
self._attr_native_value = self._get_timestamp(value)
self.async_write_ha_state()
class TeslemetryVehicleTimeSensorEntity(TeslemetryVehicleEntity, SensorEntity):

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

@ -4499,6 +4499,9 @@
}),
'name': None,
'options': dict({
'sensor': dict({
'suggested_display_precision': 1,
}),
}),
'original_device_class': <SensorDeviceClass.BATTERY: 'battery'>,
'original_icon': None,