Add typing for sense component (#129119)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Keilin Bickar 2024-10-25 04:31:33 -04:00 committed by GitHub
parent 3512cb9599
commit 8ce68f93ea
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 78 additions and 62 deletions

View File

@ -43,15 +43,15 @@ type SenseConfigEntry = ConfigEntry[SenseData]
class SenseDevicesData:
"""Data for each sense device."""
def __init__(self):
def __init__(self) -> None:
"""Create."""
self._data_by_device = {}
self._data_by_device: dict[str, dict[str, Any]] = {}
def set_devices_data(self, devices):
def set_devices_data(self, devices: list[dict[str, Any]]) -> None:
"""Store a device update."""
self._data_by_device = {device["id"]: device for device in devices}
def get_device_by_id(self, sense_device_id):
def get_device_by_id(self, sense_device_id: str) -> dict[str, Any] | None:
"""Get the latest device data."""
return self._data_by_device.get(sense_device_id)
@ -117,7 +117,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: SenseConfigEntry) -> boo
except SENSE_WEBSOCKET_EXCEPTIONS as err:
raise ConfigEntryNotReady(str(err) or "Error during realtime update") from err
async def _async_update_trend():
async def _async_update_trend() -> None:
"""Update the trend data."""
try:
await gateway.update_trend_data()
@ -156,7 +156,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: SenseConfigEntry) -> boo
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
async def async_sense_update(_):
async def async_sense_update(_) -> None:
"""Retrieve latest state."""
try:
await gateway.update_realtime()
@ -175,7 +175,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: SenseConfigEntry) -> boo
)
@callback
def _remove_update_callback_at_stop(event):
def _remove_update_callback_at_stop(event) -> None:
remove_update_callback()
entry.async_on_unload(remove_update_callback)

View File

@ -11,7 +11,7 @@ from homeassistant.helpers import entity_registry as er
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import SenseConfigEntry
from . import SenseConfigEntry, SenseDevicesData
from .const import ATTRIBUTION, DOMAIN, MDI_ICONS, SENSE_DEVICE_UPDATE
_LOGGER = logging.getLogger(__name__)
@ -38,23 +38,7 @@ async def async_setup_entry(
async_add_entities(devices)
async def _migrate_old_unique_ids(hass, devices):
registry = er.async_get(hass)
for device in devices:
# Migration of old not so unique ids
old_entity_id = registry.async_get_entity_id(
"binary_sensor", DOMAIN, device.old_unique_id
)
if old_entity_id is not None:
_LOGGER.debug(
"Migrating unique_id from [%s] to [%s]",
device.old_unique_id,
device.unique_id,
)
registry.async_update_entity(old_entity_id, new_unique_id=device.unique_id)
def sense_to_mdi(sense_icon):
def sense_to_mdi(sense_icon: str) -> str:
"""Convert sense icon to mdi icon."""
return f"mdi:{MDI_ICONS.get(sense_icon, "power-plug")}"
@ -67,7 +51,9 @@ class SenseDevice(BinarySensorEntity):
_attr_available = False
_attr_device_class = BinarySensorDeviceClass.POWER
def __init__(self, sense_devices_data, device, sense_monitor_id):
def __init__(
self, sense_devices_data: SenseDevicesData, device: dict, sense_monitor_id: str
) -> None:
"""Initialize the Sense binary sensor."""
self._attr_name = device["name"]
self._id = device["id"]
@ -77,7 +63,7 @@ class SenseDevice(BinarySensorEntity):
self._sense_devices_data = sense_devices_data
@property
def old_unique_id(self):
def old_unique_id(self) -> str:
"""Return the old not so unique id of the binary sensor."""
return self._id
@ -92,7 +78,7 @@ class SenseDevice(BinarySensorEntity):
)
@callback
def _async_update_from_data(self):
def _async_update_from_data(self) -> None:
"""Get the latest data, update state. Must not do I/O."""
new_state = bool(self._sense_devices_data.get_device_by_id(self._id))
if self._attr_available and self._attr_is_on == new_state:
@ -100,3 +86,22 @@ class SenseDevice(BinarySensorEntity):
self._attr_available = True
self._attr_is_on = new_state
self.async_write_ha_state()
async def _migrate_old_unique_ids(
hass: HomeAssistant, devices: list[SenseDevice]
) -> None:
registry = er.async_get(hass)
for device in devices:
# Migration of old not so unique ids
old_entity_id = registry.async_get_entity_id(
"binary_sensor", DOMAIN, device.old_unique_id
)
updated_id = device.unique_id
if old_entity_id is not None and updated_id is not None:
_LOGGER.debug(
"Migrating unique_id from [%s] to [%s]",
device.old_unique_id,
device.unique_id,
)
registry.async_update_entity(old_entity_id, new_unique_id=updated_id)

View File

@ -1,5 +1,10 @@
"""Support for monitoring a Sense energy sensor."""
from datetime import datetime
from typing import Any
from sense_energy import ASyncSenseable
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -15,9 +20,12 @@ from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from . import SenseConfigEntry
from . import SenseConfigEntry, SenseDevicesData
from .const import (
ACTIVE_NAME,
ACTIVE_TYPE,
@ -45,7 +53,7 @@ from .const import (
class SensorConfig:
"""Data structure holding sensor configuration."""
def __init__(self, name, sensor_type):
def __init__(self, name: str, sensor_type: str) -> None:
"""Sensor name and type to pass to API."""
self.name = name
self.sensor_type = sensor_type
@ -76,7 +84,7 @@ TREND_SENSOR_VARIANTS = [
]
def sense_to_mdi(sense_icon):
def sense_to_mdi(sense_icon: str) -> str:
"""Convert sense icon to mdi icon."""
return f"mdi:{MDI_ICONS.get(sense_icon, 'power-plug')}"
@ -160,14 +168,14 @@ class SenseActiveSensor(SensorEntity):
def __init__(
self,
data,
name,
sensor_type,
sense_monitor_id,
variant_id,
variant_name,
unique_id,
):
data: ASyncSenseable,
name: str,
sensor_type: str,
sense_monitor_id: str,
variant_id: str,
variant_name: str,
unique_id: str,
) -> None:
"""Initialize the Sense sensor."""
self._attr_name = f"{name} {variant_name}"
self._attr_unique_id = unique_id
@ -188,7 +196,7 @@ class SenseActiveSensor(SensorEntity):
)
@callback
def _async_update_from_data(self):
def _async_update_from_data(self) -> None:
"""Update the sensor from the data. Must not do I/O."""
new_state = round(
self._data.active_solar_power
@ -214,10 +222,10 @@ class SenseVoltageSensor(SensorEntity):
def __init__(
self,
data,
index,
sense_monitor_id,
):
data: ASyncSenseable,
index: int,
sense_monitor_id: str,
) -> None:
"""Initialize the Sense sensor."""
line_num = index + 1
self._attr_name = f"L{line_num} Voltage"
@ -237,7 +245,7 @@ class SenseVoltageSensor(SensorEntity):
)
@callback
def _async_update_from_data(self):
def _async_update_from_data(self) -> None:
"""Update the sensor from the data. Must not do I/O."""
new_state = round(self._data.active_voltage[self._voltage_index], 1)
if self._attr_available and self._attr_native_value == new_state:
@ -250,23 +258,20 @@ class SenseVoltageSensor(SensorEntity):
class SenseTrendsSensor(CoordinatorEntity, SensorEntity):
"""Implementation of a Sense energy sensor."""
_attr_device_class = SensorDeviceClass.ENERGY
_attr_state_class = SensorStateClass.TOTAL
_attr_native_unit_of_measurement = UnitOfEnergy.KILO_WATT_HOUR
_attr_attribution = ATTRIBUTION
_attr_should_poll = False
def __init__(
self,
data,
name,
sensor_type,
variant_id,
variant_name,
trends_coordinator,
unique_id,
sense_monitor_id,
):
data: ASyncSenseable,
name: str,
sensor_type: str,
variant_id: str,
variant_name: str,
trends_coordinator: DataUpdateCoordinator[Any],
unique_id: str,
sense_monitor_id: str,
) -> None:
"""Initialize the Sense sensor."""
super().__init__(trends_coordinator)
self._attr_name = f"{name} {variant_name}"
@ -280,6 +285,10 @@ class SenseTrendsSensor(CoordinatorEntity, SensorEntity):
self._attr_entity_registry_enabled_default = False
self._attr_state_class = None
self._attr_device_class = None
else:
self._attr_device_class = SensorDeviceClass.ENERGY
self._attr_state_class = SensorStateClass.TOTAL
self._attr_native_unit_of_measurement = UnitOfEnergy.KILO_WATT_HOUR
self._attr_device_info = DeviceInfo(
name=f"Sense {sense_monitor_id}",
identifiers={(DOMAIN, sense_monitor_id)},
@ -289,12 +298,12 @@ class SenseTrendsSensor(CoordinatorEntity, SensorEntity):
)
@property
def native_value(self):
def native_value(self) -> float:
"""Return the state of the sensor."""
return round(self._data.get_trend(self._sensor_type, self._variant_id), 1)
@property
def last_reset(self):
def last_reset(self) -> datetime | None:
"""Return the time when the sensor was last reset, if any."""
if self._attr_state_class == SensorStateClass.TOTAL:
return self._data.trend_start(self._sensor_type)
@ -311,7 +320,9 @@ class SenseEnergyDevice(SensorEntity):
_attr_device_class = SensorDeviceClass.POWER
_attr_should_poll = False
def __init__(self, sense_devices_data, device, sense_monitor_id):
def __init__(
self, sense_devices_data: SenseDevicesData, device: dict, sense_monitor_id: str
) -> None:
"""Initialize the Sense binary sensor."""
self._attr_name = f"{device['name']} {CONSUMPTION_NAME}"
self._id = device["id"]
@ -331,7 +342,7 @@ class SenseEnergyDevice(SensorEntity):
)
@callback
def _async_update_from_data(self):
def _async_update_from_data(self) -> None:
"""Get the latest data, update state. Must not do I/O."""
device_data = self._sense_devices_data.get_device_by_id(self._id)
if not device_data or "w" not in device_data: