Move idasen_desk coordinator to separate module (#117485)

This commit is contained in:
epenet 2024-05-15 11:26:40 +02:00 committed by GitHub
parent 48c03a6564
commit 6116caa7ed
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 87 additions and 73 deletions

View file

@ -2,12 +2,10 @@
from __future__ import annotations
import asyncio
import logging
from attr import dataclass
from bleak.exc import BleakError
from idasen_ha import Desk
from idasen_ha.errors import AuthFailedError
from homeassistant.components import bluetooth
@ -23,84 +21,15 @@ 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 homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN
from .coordinator import IdasenDeskCoordinator
PLATFORMS: list[Platform] = [Platform.BUTTON, Platform.COVER, Platform.SENSOR]
_LOGGER = logging.getLogger(__name__)
class IdasenDeskCoordinator(DataUpdateCoordinator[int | None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage updates for the Idasen Desk."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
name: str,
address: str,
) -> None:
"""Init IdasenDeskCoordinator."""
super().__init__(hass, logger, name=name)
self._address = address
self._expected_connected = False
self._connection_lost = False
self._disconnect_lock = asyncio.Lock()
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)
ble_device = bluetooth.async_ble_device_from_address(
self.hass, self._address, connectable=True
)
if ble_device is None:
_LOGGER.debug("No BLEDevice for %s", self._address)
return False
self._expected_connected = True
await self.desk.connect(ble_device)
return True
async def async_disconnect(self) -> None:
"""Disconnect from desk."""
_LOGGER.debug("Disconnecting from %s", self._address)
self._expected_connected = False
self._connection_lost = False
await self.desk.disconnect()
async def async_ensure_connection_state(self) -> None:
"""Check if the expected connection state matches the current state.
If the expected and current state don't match, calls connect/disconnect
as needed.
"""
if self._expected_connected:
if not self.desk.is_connected:
_LOGGER.debug("Desk disconnected. Reconnecting")
self._connection_lost = True
await self.async_connect()
elif self._connection_lost:
_LOGGER.info("Reconnected to desk")
self._connection_lost = False
elif self.desk.is_connected:
if self._disconnect_lock.locked():
_LOGGER.debug("Already disconnecting")
return
async with self._disconnect_lock:
_LOGGER.debug("Desk is connected but should not be. Disconnecting")
await self.desk.disconnect()
@callback
def async_set_updated_data(self, data: int | None) -> None:
"""Handle data update."""
self.hass.async_create_task(self.async_ensure_connection_state())
return super().async_set_updated_data(data)
@dataclass
class DeskData:
"""Data for the Idasen Desk integration."""

View file

@ -0,0 +1,83 @@
"""Coordinator for the IKEA Idasen Desk integration."""
from __future__ import annotations
import asyncio
import logging
from idasen_ha import Desk
from homeassistant.components import bluetooth
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
class IdasenDeskCoordinator(DataUpdateCoordinator[int | None]):
"""Class to manage updates for the Idasen Desk."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
name: str,
address: str,
) -> None:
"""Init IdasenDeskCoordinator."""
super().__init__(hass, logger, name=name)
self._address = address
self._expected_connected = False
self._connection_lost = False
self._disconnect_lock = asyncio.Lock()
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)
ble_device = bluetooth.async_ble_device_from_address(
self.hass, self._address, connectable=True
)
if ble_device is None:
_LOGGER.debug("No BLEDevice for %s", self._address)
return False
self._expected_connected = True
await self.desk.connect(ble_device)
return True
async def async_disconnect(self) -> None:
"""Disconnect from desk."""
_LOGGER.debug("Disconnecting from %s", self._address)
self._expected_connected = False
self._connection_lost = False
await self.desk.disconnect()
async def async_ensure_connection_state(self) -> None:
"""Check if the expected connection state matches the current state.
If the expected and current state don't match, calls connect/disconnect
as needed.
"""
if self._expected_connected:
if not self.desk.is_connected:
_LOGGER.debug("Desk disconnected. Reconnecting")
self._connection_lost = True
await self.async_connect()
elif self._connection_lost:
_LOGGER.info("Reconnected to desk")
self._connection_lost = False
elif self.desk.is_connected:
if self._disconnect_lock.locked():
_LOGGER.debug("Already disconnecting")
return
async with self._disconnect_lock:
_LOGGER.debug("Desk is connected but should not be. Disconnecting")
await self.desk.disconnect()
@callback
def async_set_updated_data(self, data: int | None) -> None:
"""Handle data update."""
self.hass.async_create_task(self.async_ensure_connection_state())
return super().async_set_updated_data(data)

View file

@ -19,7 +19,9 @@ def mock_bluetooth(enable_bluetooth):
@pytest.fixture(autouse=False)
def mock_desk_api():
"""Set up idasen desk API fixture."""
with mock.patch("homeassistant.components.idasen_desk.Desk") as desk_patched:
with mock.patch(
"homeassistant.components.idasen_desk.coordinator.Desk"
) as desk_patched:
mock_desk = MagicMock()
def mock_init(