Use a common base entity for Idasen Desk (#132496)

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
pull/133470/head
Abílio Costa 2024-12-18 09:20:14 +00:00 committed by GitHub
parent 413a578fdb
commit 5fb5e933e2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 82 additions and 118 deletions

View File

@ -4,53 +4,31 @@ from __future__ import annotations
import logging
from attr import dataclass
from bleak.exc import BleakError
from idasen_ha.errors import AuthFailedError
from homeassistant.components import bluetooth
from homeassistant.components.bluetooth.match import ADDRESS, BluetoothCallbackMatcher
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import (
ATTR_NAME,
CONF_ADDRESS,
EVENT_HOMEASSISTANT_STOP,
Platform,
)
from homeassistant.const import CONF_ADDRESS, EVENT_HOMEASSISTANT_STOP, Platform
from homeassistant.core import Event, HomeAssistant, callback
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.device_registry import DeviceInfo
from .const import DOMAIN
from .coordinator import IdasenDeskCoordinator
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.COVER, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
@dataclass
class DeskData:
"""Data for the Idasen Desk integration."""
address: str
device_info: DeviceInfo
coordinator: IdasenDeskCoordinator
type IdasenDeskConfigEntry = ConfigEntry[IdasenDeskCoordinator]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(hass: HomeAssistant, entry: IdasenDeskConfigEntry) -> bool:
"""Set up IKEA Idasen from a config entry."""
address: str = entry.data[CONF_ADDRESS].upper()
coordinator = IdasenDeskCoordinator(hass, _LOGGER, entry.title, address)
device_info = DeviceInfo(
name=entry.title,
connections={(dr.CONNECTION_BLUETOOTH, address)},
)
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = DeskData(
address, device_info, coordinator
)
entry.runtime_data = coordinator
try:
if not await coordinator.async_connect():
@ -89,18 +67,18 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return True
async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> None:
async def _async_update_listener(
hass: HomeAssistant, entry: IdasenDeskConfigEntry
) -> None:
"""Handle options update."""
data: DeskData = hass.data[DOMAIN][entry.entry_id]
if entry.title != data.device_info[ATTR_NAME]:
await hass.config_entries.async_reload(entry.entry_id)
await hass.config_entries.async_reload(entry.entry_id)
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(hass: HomeAssistant, entry: IdasenDeskConfigEntry) -> bool:
"""Unload a config entry."""
if unload_ok := await hass.config_entries.async_unload_platforms(entry, PLATFORMS):
data: DeskData = hass.data[DOMAIN].pop(entry.entry_id)
await data.coordinator.async_disconnect()
bluetooth.async_rediscover_address(hass, data.address)
coordinator = entry.runtime_data
await coordinator.async_disconnect()
bluetooth.async_rediscover_address(hass, coordinator.address)
return unload_ok

View File

@ -6,14 +6,12 @@ import logging
from typing import Any, Final
from homeassistant.components.button import ButtonEntity, ButtonEntityDescription
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import DeskData, IdasenDeskCoordinator
from .const import DOMAIN
from . import IdasenDeskConfigEntry, IdasenDeskCoordinator
from .entity import IdasenDeskEntity
_LOGGER = logging.getLogger(__name__)
@ -45,43 +43,38 @@ BUTTONS: Final = [
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: IdasenDeskConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set buttons for device."""
data: DeskData = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
IdasenDeskButton(data.address, data.device_info, data.coordinator, button)
for button in BUTTONS
)
coordinator = entry.runtime_data
async_add_entities(IdasenDeskButton(coordinator, button) for button in BUTTONS)
class IdasenDeskButton(ButtonEntity):
class IdasenDeskButton(IdasenDeskEntity, ButtonEntity):
"""Defines a IdasenDesk button."""
entity_description: IdasenDeskButtonDescription
_attr_has_entity_name = True
def __init__(
self,
address: str,
device_info: DeviceInfo,
coordinator: IdasenDeskCoordinator,
description: IdasenDeskButtonDescription,
) -> None:
"""Initialize the IdasenDesk button entity."""
super().__init__(f"{description.key}-{coordinator.address}", coordinator)
self.entity_description = description
self._attr_unique_id = f"{description.key}-{address}"
self._attr_device_info = device_info
self._address = address
self._coordinator = coordinator
async def async_press(self) -> None:
"""Triggers the IdasenDesk button press service."""
_LOGGER.debug(
"Trigger %s for %s",
self.entity_description.key,
self._address,
self.coordinator.address,
)
await self.entity_description.press_action(self._coordinator)()
await self.entity_description.press_action(self.coordinator)()
@property
def available(self) -> bool:
"""Connect/disconnect buttons should always be available."""
return True

View File

@ -26,20 +26,20 @@ class IdasenDeskCoordinator(DataUpdateCoordinator[int | None]):
"""Init IdasenDeskCoordinator."""
super().__init__(hass, logger, name=name)
self._address = address
self.address = address
self._expected_connected = False
self.desk = Desk(self.async_set_updated_data)
async def async_connect(self) -> bool:
"""Connect to desk."""
_LOGGER.debug("Trying to connect %s", self._address)
_LOGGER.debug("Trying to connect %s", self.address)
self._expected_connected = True
ble_device = bluetooth.async_ble_device_from_address(
self.hass, self._address, connectable=True
self.hass, self.address, connectable=True
)
if ble_device is None:
_LOGGER.debug("No BLEDevice for %s", self._address)
_LOGGER.debug("No BLEDevice for %s", self.address)
return False
await self.desk.connect(ble_device)
return True
@ -47,7 +47,7 @@ class IdasenDeskCoordinator(DataUpdateCoordinator[int | None]):
async def async_disconnect(self) -> None:
"""Disconnect from desk."""
self._expected_connected = False
_LOGGER.debug("Disconnecting from %s", self._address)
_LOGGER.debug("Disconnecting from %s", self.address)
await self.desk.disconnect()
async def async_connect_if_expected(self) -> None:

View File

@ -12,30 +12,25 @@ from homeassistant.components.cover import (
CoverEntity,
CoverEntityFeature,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DeskData, IdasenDeskCoordinator
from .const import DOMAIN
from . import IdasenDeskConfigEntry, IdasenDeskCoordinator
from .entity import IdasenDeskEntity
async def async_setup_entry(
hass: HomeAssistant,
entry: ConfigEntry,
entry: IdasenDeskConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the cover platform for Idasen Desk."""
data: DeskData = hass.data[DOMAIN][entry.entry_id]
async_add_entities(
[IdasenDeskCover(data.address, data.device_info, data.coordinator)]
)
coordinator = entry.runtime_data
async_add_entities([IdasenDeskCover(coordinator)])
class IdasenDeskCover(CoordinatorEntity[IdasenDeskCoordinator], CoverEntity):
class IdasenDeskCover(IdasenDeskEntity, CoverEntity):
"""Representation of Idasen Desk device."""
_attr_device_class = CoverDeviceClass.DAMPER
@ -45,29 +40,14 @@ class IdasenDeskCover(CoordinatorEntity[IdasenDeskCoordinator], CoverEntity):
| CoverEntityFeature.STOP
| CoverEntityFeature.SET_POSITION
)
_attr_has_entity_name = True
_attr_name = None
_attr_translation_key = "desk"
def __init__(
self,
address: str,
device_info: DeviceInfo,
coordinator: IdasenDeskCoordinator,
) -> None:
def __init__(self, coordinator: IdasenDeskCoordinator) -> None:
"""Initialize an Idasen Desk cover."""
super().__init__(coordinator)
self._desk = coordinator.desk
self._attr_unique_id = address
self._attr_device_info = device_info
super().__init__(coordinator.address, coordinator)
self._attr_current_cover_position = self._desk.height_percent
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self._desk.is_connected is True
@property
def is_closed(self) -> bool:
"""Return if the cover is closed."""

View File

@ -0,0 +1,34 @@
"""Base entity for Idasen Desk."""
from __future__ import annotations
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import IdasenDeskCoordinator
class IdasenDeskEntity(CoordinatorEntity[IdasenDeskCoordinator]):
"""IdasenDesk sensor."""
_attr_has_entity_name = True
def __init__(
self,
unique_id: str,
coordinator: IdasenDeskCoordinator,
) -> None:
"""Initialize the IdasenDesk sensor entity."""
super().__init__(coordinator)
self._attr_unique_id = unique_id
self._attr_device_info = dr.DeviceInfo(
manufacturer="LINAK",
connections={(dr.CONNECTION_BLUETOOTH, coordinator.address)},
)
self._desk = coordinator.desk
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self._desk.is_connected is True

View File

@ -9,10 +9,7 @@ rules:
comment: |
This integration does not use polling.
brands: done
common-modules:
status: todo
comment: |
The cover and sensor entities could move common initialization to a base entity class.
common-modules: done
config-flow-test-coverage:
status: todo
comment: |
@ -33,7 +30,7 @@ rules:
entity-event-setup: done
entity-unique-id: done
has-entity-name: done
runtime-data: todo
runtime-data: done
test-before-configure: done
test-before-setup: done
unique-config-entry: done

View File

@ -6,7 +6,6 @@ from collections.abc import Callable
from dataclasses import dataclass
from typing import Any
from homeassistant import config_entries
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -15,12 +14,10 @@ from homeassistant.components.sensor import (
)
from homeassistant.const import UnitOfLength
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import DeskData, IdasenDeskCoordinator
from .const import DOMAIN
from . import IdasenDeskConfigEntry, IdasenDeskCoordinator
from .entity import IdasenDeskEntity
@dataclass(frozen=True, kw_only=True)
@ -46,51 +43,36 @@ SENSORS = (
async def async_setup_entry(
hass: HomeAssistant,
entry: config_entries.ConfigEntry,
entry: IdasenDeskConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Idasen Desk sensors."""
data: DeskData = hass.data[DOMAIN][entry.entry_id]
coordinator = entry.runtime_data
async_add_entities(
IdasenDeskSensor(
data.address, data.device_info, data.coordinator, sensor_description
)
IdasenDeskSensor(coordinator, sensor_description)
for sensor_description in SENSORS
)
class IdasenDeskSensor(CoordinatorEntity[IdasenDeskCoordinator], SensorEntity):
class IdasenDeskSensor(IdasenDeskEntity, SensorEntity):
"""IdasenDesk sensor."""
entity_description: IdasenDeskSensorDescription
_attr_has_entity_name = True
def __init__(
self,
address: str,
device_info: DeviceInfo,
coordinator: IdasenDeskCoordinator,
description: IdasenDeskSensorDescription,
) -> None:
"""Initialize the IdasenDesk sensor entity."""
super().__init__(coordinator)
super().__init__(f"{description.key}-{coordinator.address}", coordinator)
self.entity_description = description
self._attr_unique_id = f"{description.key}-{address}"
self._attr_device_info = device_info
self._address = address
self._desk = coordinator.desk
async def async_added_to_hass(self) -> None:
"""When entity is added to hass."""
await super().async_added_to_hass()
self._update_native_value()
@property
def available(self) -> bool:
"""Return True if entity is available."""
return super().available and self._desk.is_connected is True
@callback
def _handle_coordinator_update(self, *args: Any) -> None:
"""Handle data update."""