From 186141ee4df74de2d0b6cb744652360e8b4cb558 Mon Sep 17 00:00:00 2001 From: epenet <6771947+epenet@users.noreply.github.com> Date: Thu, 23 Jun 2022 21:35:05 +0200 Subject: [PATCH] Use attributes in keba locks and binary sensors (#73894) Co-authored-by: Franck Nijhof --- .../components/keba/binary_sensor.py | 70 ++++++++----------- homeassistant/components/keba/lock.py | 46 ++++-------- homeassistant/components/keba/sensor.py | 8 +-- 3 files changed, 46 insertions(+), 78 deletions(-) diff --git a/homeassistant/components/keba/binary_sensor.py b/homeassistant/components/keba/binary_sensor.py index 19f3bd428ec..7997130c90a 100644 --- a/homeassistant/components/keba/binary_sensor.py +++ b/homeassistant/components/keba/binary_sensor.py @@ -1,6 +1,8 @@ """Support for KEBA charging station binary sensors.""" from __future__ import annotations +from typing import Any + from homeassistant.components.binary_sensor import ( BinarySensorDeviceClass, BinarySensorEntity, @@ -9,7 +11,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -from . import DOMAIN +from . import DOMAIN, KebaHandler async def async_setup_platform( @@ -22,7 +24,7 @@ async def async_setup_platform( if discovery_info is None: return - keba = hass.data[DOMAIN] + keba: KebaHandler = hass.data[DOMAIN] sensors = [ KebaBinarySensor( @@ -60,53 +62,37 @@ async def async_setup_platform( class KebaBinarySensor(BinarySensorEntity): """Representation of a binary sensor of a KEBA charging station.""" - def __init__(self, keba, key, name, entity_type, device_class): + _attr_should_poll = False + + def __init__( + self, + keba: KebaHandler, + key: str, + name: str, + entity_type: str, + device_class: BinarySensorDeviceClass, + ) -> None: """Initialize the KEBA Sensor.""" self._key = key self._keba = keba - self._name = name - self._entity_type = entity_type - self._device_class = device_class - self._is_on = None - self._attributes = {} + self._attributes: dict[str, Any] = {} + + self._attr_device_class = device_class + self._attr_name = f"{keba.device_name} {name}" + self._attr_unique_id = f"{keba.device_id}_{entity_type}" @property - def should_poll(self): - """Deactivate polling. Data updated by KebaHandler.""" - return False - - @property - def unique_id(self): - """Return the unique ID of the binary sensor.""" - return f"{self._keba.device_id}_{self._entity_type}" - - @property - def name(self): - """Return the name of the device.""" - return f"{self._keba.device_name} {self._name}" - - @property - def device_class(self): - """Return the class of this sensor.""" - return self._device_class - - @property - def is_on(self): - """Return true if sensor is on.""" - return self._is_on - - @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, Any]: """Return the state attributes of the binary sensor.""" return self._attributes - async def async_update(self): + async def async_update(self) -> None: """Get latest cached states from the device.""" if self._key == "Online": - self._is_on = self._keba.get_value(self._key) + self._attr_is_on = self._keba.get_value(self._key) elif self._key == "Plug": - self._is_on = self._keba.get_value("Plug_plugged") + self._attr_is_on = self._keba.get_value("Plug_plugged") self._attributes["plugged_on_wallbox"] = self._keba.get_value( "Plug_wallbox" ) @@ -114,23 +100,23 @@ class KebaBinarySensor(BinarySensorEntity): self._attributes["plugged_on_EV"] = self._keba.get_value("Plug_EV") elif self._key == "State": - self._is_on = self._keba.get_value("State_on") + self._attr_is_on = self._keba.get_value("State_on") self._attributes["status"] = self._keba.get_value("State_details") self._attributes["max_charging_rate"] = str( self._keba.get_value("Max curr") ) elif self._key == "Tmo FS": - self._is_on = not self._keba.get_value("FS_on") + self._attr_is_on = not self._keba.get_value("FS_on") self._attributes["failsafe_timeout"] = str(self._keba.get_value("Tmo FS")) self._attributes["fallback_current"] = str(self._keba.get_value("Curr FS")) elif self._key == "Authreq": - self._is_on = self._keba.get_value(self._key) == 0 + self._attr_is_on = self._keba.get_value(self._key) == 0 - def update_callback(self): + def update_callback(self) -> None: """Schedule a state update.""" self.async_schedule_update_ha_state(True) - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Add update callback after being added to hass.""" self._keba.add_update_listener(self.update_callback) diff --git a/homeassistant/components/keba/lock.py b/homeassistant/components/keba/lock.py index d0316b1e525..de8d28d7739 100644 --- a/homeassistant/components/keba/lock.py +++ b/homeassistant/components/keba/lock.py @@ -8,7 +8,7 @@ from homeassistant.core import HomeAssistant from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType -from . import DOMAIN +from . import DOMAIN, KebaHandler async def async_setup_platform( @@ -21,41 +21,23 @@ async def async_setup_platform( if discovery_info is None: return - keba = hass.data[DOMAIN] + keba: KebaHandler = hass.data[DOMAIN] - sensors = [KebaLock(keba, "Authentication", "authentication")] - async_add_entities(sensors) + locks = [KebaLock(keba, "Authentication", "authentication")] + async_add_entities(locks) class KebaLock(LockEntity): """The entity class for KEBA charging stations switch.""" - def __init__(self, keba, name, entity_type): + _attr_should_poll = False + + def __init__(self, keba: KebaHandler, name: str, entity_type: str) -> None: """Initialize the KEBA switch.""" self._keba = keba - self._name = name - self._entity_type = entity_type - self._state = True - - @property - def should_poll(self): - """Deactivate polling. Data updated by KebaHandler.""" - return False - - @property - def unique_id(self): - """Return the unique ID of the lock.""" - return f"{self._keba.device_id}_{self._entity_type}" - - @property - def name(self): - """Return the name of the device.""" - return f"{self._keba.device_name} {self._name}" - - @property - def is_locked(self) -> bool: - """Return true if lock is locked.""" - return self._state + self._attr_is_locked = True + self._attr_name = f"{keba.device_name} {name}" + self._attr_unique_id = f"{keba.device_id}_{entity_type}" async def async_lock(self, **kwargs: Any) -> None: """Lock wallbox.""" @@ -65,14 +47,14 @@ class KebaLock(LockEntity): """Unlock wallbox.""" await self._keba.async_start() - async def async_update(self): + async def async_update(self) -> None: """Attempt to retrieve on off state from the switch.""" - self._state = self._keba.get_value("Authreq") == 1 + self._attr_is_locked = self._keba.get_value("Authreq") == 1 - def update_callback(self): + def update_callback(self) -> None: """Schedule a state update.""" self.async_schedule_update_ha_state(True) - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Add update callback after being added to hass.""" self._keba.add_update_listener(self.update_callback) diff --git a/homeassistant/components/keba/sensor.py b/homeassistant/components/keba/sensor.py index 6309f0b79fb..d35c22905f1 100644 --- a/homeassistant/components/keba/sensor.py +++ b/homeassistant/components/keba/sensor.py @@ -109,11 +109,11 @@ class KebaSensor(SensorEntity): self._attributes: dict[str, str] = {} @property - def extra_state_attributes(self): + def extra_state_attributes(self) -> dict[str, str]: """Return the state attributes of the binary sensor.""" return self._attributes - async def async_update(self): + async def async_update(self) -> None: """Get latest cached states from the device.""" self._attr_native_value = self._keba.get_value(self.entity_description.key) @@ -128,10 +128,10 @@ class KebaSensor(SensorEntity): elif self.entity_description.key == "Curr user": self._attributes["max_current_hardware"] = self._keba.get_value("Curr HW") - def update_callback(self): + def update_callback(self) -> None: """Schedule a state update.""" self.async_schedule_update_ha_state(True) - async def async_added_to_hass(self): + async def async_added_to_hass(self) -> None: """Add update callback after being added to hass.""" self._keba.add_update_listener(self.update_callback)