diff --git a/homeassistant/components/sense/__init__.py b/homeassistant/components/sense/__init__.py
index 58e993ad6e0c247368db3e8cedf592ee7a987ecc..ea424798891b3f4b4d6f4e74cec1c5bd6e3b96ad 100644
--- a/homeassistant/components/sense/__init__.py
+++ b/homeassistant/components/sense/__init__.py
@@ -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)
diff --git a/homeassistant/components/sense/binary_sensor.py b/homeassistant/components/sense/binary_sensor.py
index 8317f8458b3449e158d68414f31d452e27c7e59c..969dfdc565e40759e432d6b7ed5705ea7f733b7b 100644
--- a/homeassistant/components/sense/binary_sensor.py
+++ b/homeassistant/components/sense/binary_sensor.py
@@ -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)
diff --git a/homeassistant/components/sense/sensor.py b/homeassistant/components/sense/sensor.py
index bc9dd470f5eed7bf6bd5157b11e1ba993432adf5..053cc39d20cee8906aaaaca88bb7da7f3f47f0d2 100644
--- a/homeassistant/components/sense/sensor.py
+++ b/homeassistant/components/sense/sensor.py
@@ -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: