diff --git a/homeassistant/components/linear_garage_door/__init__.py b/homeassistant/components/linear_garage_door/__init__.py index e21d8eaba58..16e743e00b5 100644 --- a/homeassistant/components/linear_garage_door/__init__.py +++ b/homeassistant/components/linear_garage_door/__init__.py @@ -15,7 +15,7 @@ PLATFORMS: list[Platform] = [Platform.COVER] async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up Linear Garage Door from a config entry.""" - coordinator = LinearUpdateCoordinator(hass, entry) + coordinator = LinearUpdateCoordinator(hass) await coordinator.async_config_entry_first_refresh() diff --git a/homeassistant/components/linear_garage_door/coordinator.py b/homeassistant/components/linear_garage_door/coordinator.py index b771b552b62..91ff0165163 100644 --- a/homeassistant/components/linear_garage_door/coordinator.py +++ b/homeassistant/components/linear_garage_door/coordinator.py @@ -2,9 +2,11 @@ from __future__ import annotations +from collections.abc import Awaitable, Callable +from dataclasses import dataclass from datetime import timedelta import logging -from typing import Any +from typing import Any, TypeVar from linear_garage_door import Linear from linear_garage_door.errors import InvalidLoginError @@ -17,46 +19,58 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator _LOGGER = logging.getLogger(__name__) +_T = TypeVar("_T") -class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): + +@dataclass +class LinearDevice: + """Linear device dataclass.""" + + name: str + subdevices: dict[str, dict[str, str]] + + +class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, LinearDevice]]): """DataUpdateCoordinator for Linear.""" - _email: str - _password: str - _device_id: str - _site_id: str - _devices: list[dict[str, list[str] | str]] | None - _linear: Linear + _devices: list[dict[str, Any]] | None = None + config_entry: ConfigEntry - def __init__( - self, - hass: HomeAssistant, - entry: ConfigEntry, - ) -> None: + def __init__(self, hass: HomeAssistant) -> None: """Initialize DataUpdateCoordinator for Linear.""" - self._email = entry.data["email"] - self._password = entry.data["password"] - self._device_id = entry.data["device_id"] - self._site_id = entry.data["site_id"] - self._devices = None - super().__init__( hass, _LOGGER, name="Linear Garage Door", update_interval=timedelta(seconds=60), ) + self.site_id = self.config_entry.data["site_id"] - async def _async_update_data(self) -> dict[str, Any]: + async def _async_update_data(self) -> dict[str, LinearDevice]: """Get the data for Linear.""" - linear = Linear() + async def update_data(linear: Linear) -> dict[str, Any]: + if not self._devices: + self._devices = await linear.get_devices(self.site_id) + data = {} + + for device in self._devices: + device_id = str(device["id"]) + state = await linear.get_device_state(device_id) + data[device_id] = LinearDevice(device["name"], state) + return data + + return await self.execute(update_data) + + async def execute(self, func: Callable[[Linear], Awaitable[_T]]) -> _T: + """Execute an API call.""" + linear = Linear() try: await linear.login( - email=self._email, - password=self._password, - device_id=self._device_id, + email=self.config_entry.data["email"], + password=self.config_entry.data["password"], + device_id=self.config_entry.data["device_id"], client_session=async_get_clientsession(self.hass), ) except InvalidLoginError as err: @@ -66,17 +80,6 @@ class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]): ): raise ConfigEntryAuthFailed from err raise ConfigEntryNotReady from err - - if not self._devices: - self._devices = await linear.get_devices(self._site_id) - - data = {} - - for device in self._devices: - device_id = str(device["id"]) - state = await linear.get_device_state(device_id) - data[device_id] = {"name": device["name"], "subdevices": state} - + result = await func(linear) await linear.close() - - return data + return result diff --git a/homeassistant/components/linear_garage_door/cover.py b/homeassistant/components/linear_garage_door/cover.py index 3474e9d3acb..b3d720e531a 100644 --- a/homeassistant/components/linear_garage_door/cover.py +++ b/homeassistant/components/linear_garage_door/cover.py @@ -3,8 +3,6 @@ from datetime import timedelta from typing import Any -from linear_garage_door import Linear - from homeassistant.components.cover import ( CoverDeviceClass, CoverEntity, @@ -12,13 +10,12 @@ from homeassistant.components.cover import ( ) from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant -from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.update_coordinator import CoordinatorEntity from .const import DOMAIN -from .coordinator import LinearUpdateCoordinator +from .coordinator import LinearDevice, LinearUpdateCoordinator SUPPORTED_SUBDEVICES = ["GDO"] PARALLEL_UPDATES = 1 @@ -32,118 +29,89 @@ async def async_setup_entry( ) -> None: """Set up Linear Garage Door cover.""" coordinator: LinearUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id] - data = coordinator.data - device_list: list[LinearCoverEntity] = [] - - for device_id in data: - device_list.extend( - LinearCoverEntity( - device_id=device_id, - device_name=data[device_id]["name"], - subdevice=subdev, - config_entry=config_entry, - coordinator=coordinator, - ) - for subdev in data[device_id]["subdevices"] - if subdev in SUPPORTED_SUBDEVICES - ) - async_add_entities(device_list) + async_add_entities( + LinearCoverEntity(coordinator, device_id, sub_device_id) + for device_id, device_data in coordinator.data.items() + for sub_device_id in device_data.subdevices + if sub_device_id in SUPPORTED_SUBDEVICES + ) class LinearCoverEntity(CoordinatorEntity[LinearUpdateCoordinator], CoverEntity): """Representation of a Linear cover.""" _attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE + _attr_has_entity_name = True + _attr_name = None + _attr_device_class = CoverDeviceClass.GARAGE def __init__( self, - device_id: str, - device_name: str, - subdevice: str, - config_entry: ConfigEntry, coordinator: LinearUpdateCoordinator, + device_id: str, + sub_device_id: str, ) -> None: """Init with device ID and name.""" super().__init__(coordinator) - - self._attr_has_entity_name = True - self._attr_name = None self._device_id = device_id - self._device_name = device_name - self._subdevice = subdevice - self._attr_device_class = CoverDeviceClass.GARAGE - self._attr_unique_id = f"{device_id}-{subdevice}" - self._config_entry = config_entry - - def _get_data(self, data_property: str) -> str: - """Get a property of the subdevice.""" - return str( - self.coordinator.data[self._device_id]["subdevices"][self._subdevice].get( - data_property - ) - ) - - @property - def device_info(self) -> DeviceInfo: - """Return device info of a garage door.""" - return DeviceInfo( - identifiers={(DOMAIN, self._device_id)}, - name=self._device_name, + self._sub_device_id = sub_device_id + self._attr_unique_id = f"{device_id}-{sub_device_id}" + self._attr_device_info = DeviceInfo( + identifiers={(DOMAIN, sub_device_id)}, + name=self.linear_device.name, manufacturer="Linear", model="Garage Door Opener", ) + @property + def linear_device(self) -> LinearDevice: + """Return the Linear device.""" + return self.coordinator.data[self._device_id] + + @property + def sub_device(self) -> dict[str, str]: + """Return the subdevice.""" + return self.linear_device.subdevices[self._sub_device_id] + @property def is_closed(self) -> bool: """Return if cover is closed.""" - return bool(self._get_data("Open_B") == "false") + return self.sub_device.get("Open_B") == "false" @property def is_opened(self) -> bool: """Return if cover is open.""" - return bool(self._get_data("Open_B") == "true") + return self.sub_device.get("Open_B") == "true" @property def is_opening(self) -> bool: """Return if cover is opening.""" - return bool(self._get_data("Opening_P") == "0") + return self.sub_device.get("Opening_P") == "0" @property def is_closing(self) -> bool: """Return if cover is closing.""" - return bool(self._get_data("Opening_P") == "100") + return self.sub_device.get("Opening_P") == "100" async def async_close_cover(self, **kwargs: Any) -> None: """Close the garage door.""" if self.is_closed: return - linear = Linear() - - await linear.login( - email=self._config_entry.data["email"], - password=self._config_entry.data["password"], - device_id=self._config_entry.data["device_id"], - client_session=async_get_clientsession(self.hass), + await self.coordinator.execute( + lambda linear: linear.operate_device( + self._device_id, self._sub_device_id, "Close" + ) ) - await linear.operate_device(self._device_id, self._subdevice, "Close") - await linear.close() - async def async_open_cover(self, **kwargs: Any) -> None: """Open the garage door.""" if self.is_opened: return - linear = Linear() - - await linear.login( - email=self._config_entry.data["email"], - password=self._config_entry.data["password"], - device_id=self._config_entry.data["device_id"], - client_session=async_get_clientsession(self.hass), + await self.coordinator.execute( + lambda linear: linear.operate_device( + self._device_id, self._sub_device_id, "Open" + ) ) - - await linear.operate_device(self._device_id, self._subdevice, "Open") - await linear.close() diff --git a/homeassistant/components/linear_garage_door/diagnostics.py b/homeassistant/components/linear_garage_door/diagnostics.py index fc4906daa77..21414f02f87 100644 --- a/homeassistant/components/linear_garage_door/diagnostics.py +++ b/homeassistant/components/linear_garage_door/diagnostics.py @@ -2,6 +2,7 @@ from __future__ import annotations +from dataclasses import asdict from typing import Any from homeassistant.components.diagnostics import async_redact_data @@ -23,5 +24,8 @@ async def async_get_config_entry_diagnostics( return { "entry": async_redact_data(entry.as_dict(), TO_REDACT), - "coordinator_data": coordinator.data, + "coordinator_data": { + device_id: asdict(device_data) + for device_id, device_data in coordinator.data.items() + }, } diff --git a/tests/components/linear_garage_door/test_cover.py b/tests/components/linear_garage_door/test_cover.py index 9db7b80fd0e..6236d2ba39c 100644 --- a/tests/components/linear_garage_door/test_cover.py +++ b/tests/components/linear_garage_door/test_cover.py @@ -45,7 +45,7 @@ async def test_open_cover(hass: HomeAssistant) -> None: await async_init_integration(hass) with patch( - "homeassistant.components.linear_garage_door.cover.Linear.operate_device" + "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device" ) as operate_device: await hass.services.async_call( COVER_DOMAIN, @@ -58,15 +58,15 @@ async def test_open_cover(hass: HomeAssistant) -> None: with ( patch( - "homeassistant.components.linear_garage_door.cover.Linear.login", + "homeassistant.components.linear_garage_door.coordinator.Linear.login", return_value=True, ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.operate_device", + "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device", return_value=None, ) as operate_device, patch( - "homeassistant.components.linear_garage_door.cover.Linear.close", + "homeassistant.components.linear_garage_door.coordinator.Linear.close", return_value=True, ), ): @@ -80,11 +80,11 @@ async def test_open_cover(hass: HomeAssistant) -> None: assert operate_device.call_count == 1 with ( patch( - "homeassistant.components.linear_garage_door.cover.Linear.login", + "homeassistant.components.linear_garage_door.coordinator.Linear.login", return_value=True, ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.get_devices", + "homeassistant.components.linear_garage_door.coordinator.Linear.get_devices", return_value=[ { "id": "test1", @@ -99,7 +99,7 @@ async def test_open_cover(hass: HomeAssistant) -> None: ], ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.get_device_state", + "homeassistant.components.linear_garage_door.coordinator.Linear.get_device_state", side_effect=lambda id: { "test1": { "GDO": {"Open_B": "true", "Open_P": "100"}, @@ -120,7 +120,7 @@ async def test_open_cover(hass: HomeAssistant) -> None: }[id], ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.close", + "homeassistant.components.linear_garage_door.coordinator.Linear.close", return_value=True, ), ): @@ -136,7 +136,7 @@ async def test_close_cover(hass: HomeAssistant) -> None: await async_init_integration(hass) with patch( - "homeassistant.components.linear_garage_door.cover.Linear.operate_device" + "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device" ) as operate_device: await hass.services.async_call( COVER_DOMAIN, @@ -149,15 +149,15 @@ async def test_close_cover(hass: HomeAssistant) -> None: with ( patch( - "homeassistant.components.linear_garage_door.cover.Linear.login", + "homeassistant.components.linear_garage_door.coordinator.Linear.login", return_value=True, ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.operate_device", + "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device", return_value=None, ) as operate_device, patch( - "homeassistant.components.linear_garage_door.cover.Linear.close", + "homeassistant.components.linear_garage_door.coordinator.Linear.close", return_value=True, ), ): @@ -171,11 +171,11 @@ async def test_close_cover(hass: HomeAssistant) -> None: assert operate_device.call_count == 1 with ( patch( - "homeassistant.components.linear_garage_door.cover.Linear.login", + "homeassistant.components.linear_garage_door.coordinator.Linear.login", return_value=True, ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.get_devices", + "homeassistant.components.linear_garage_door.coordinator.Linear.get_devices", return_value=[ { "id": "test1", @@ -190,7 +190,7 @@ async def test_close_cover(hass: HomeAssistant) -> None: ], ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.get_device_state", + "homeassistant.components.linear_garage_door.coordinator.Linear.get_device_state", side_effect=lambda id: { "test1": { "GDO": {"Open_B": "true", "Opening_P": "100"}, @@ -211,7 +211,7 @@ async def test_close_cover(hass: HomeAssistant) -> None: }[id], ), patch( - "homeassistant.components.linear_garage_door.cover.Linear.close", + "homeassistant.components.linear_garage_door.coordinator.Linear.close", return_value=True, ), ):