jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-26 22:57:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve linear coordinator (#116167)

Browse Source 
* Improve linear coordinator

* Fix

* Fix
This commit is contained in:
Joost Lekkerkerker 2024-04-25 22:57:29 +02:00 committed by GitHub
parent ccc2f6c5b5
commit 4a1e1bd1b9
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 101 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/linear_garage_door/__init__.py
View File

@ -15,7 +15,7 @@ PLATFORMS: list[Platform] = [Platform.COVER]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
"""Set up Linear Garage Door from a config entry.""" """Set up Linear Garage Door from a config entry."""
coordinator = LinearUpdateCoordinator(hass, entry) coordinator = LinearUpdateCoordinator(hass)
await coordinator.async_config_entry_first_refresh() await coordinator.async_config_entry_first_refresh()

77
homeassistant/components/linear_garage_door/coordinator.py
View File

@ -2,9 +2,11 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from datetime import timedelta from datetime import timedelta
import logging import logging
from typing import Any from typing import Any, TypeVar
from linear_garage_door import Linear from linear_garage_door import Linear
from linear_garage_door.errors import InvalidLoginError from linear_garage_door.errors import InvalidLoginError
@ -17,46 +19,58 @@ from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
_LOGGER = logging.getLogger(__name__) _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.""" """DataUpdateCoordinator for Linear."""
_email: str _devices: list[dict[str, Any]] | None = None
_password: str config_entry: ConfigEntry
_device_id: str
_site_id: str
_devices: list[dict[str, list[str] | str]] | None
_linear: Linear
def __init__( def __init__(self, hass: HomeAssistant) -> None:
self,
hass: HomeAssistant,
entry: ConfigEntry,
) -> None:
"""Initialize DataUpdateCoordinator for Linear.""" """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__( super().__init__(
hass, hass,
_LOGGER, _LOGGER,
name="Linear Garage Door", name="Linear Garage Door",
update_interval=timedelta(seconds=60), 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.""" """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: try:
await linear.login( await linear.login(
email=self._email, email=self.config_entry.data["email"],
password=self._password, password=self.config_entry.data["password"],
device_id=self._device_id, device_id=self.config_entry.data["device_id"],
client_session=async_get_clientsession(self.hass), client_session=async_get_clientsession(self.hass),
) )
except InvalidLoginError as err: except InvalidLoginError as err:
@ -66,17 +80,6 @@ class LinearUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
): ):
raise ConfigEntryAuthFailed from err raise ConfigEntryAuthFailed from err
raise ConfigEntryNotReady from err raise ConfigEntryNotReady from err
result = await func(linear)
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}
await linear.close() await linear.close()
return result
return data

108
homeassistant/components/linear_garage_door/cover.py
View File

@ -3,8 +3,6 @@
from datetime import timedelta from datetime import timedelta
from typing import Any from typing import Any
from linear_garage_door import Linear
from homeassistant.components.cover import ( from homeassistant.components.cover import (
CoverDeviceClass, CoverDeviceClass,
CoverEntity, CoverEntity,
@ -12,13 +10,12 @@ from homeassistant.components.cover import (
) )
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant from homeassistant.core import HomeAssistant
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.device_registry import DeviceInfo from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN from .const import DOMAIN
from .coordinator import LinearUpdateCoordinator from .coordinator import LinearDevice, LinearUpdateCoordinator
SUPPORTED_SUBDEVICES = ["GDO"] SUPPORTED_SUBDEVICES = ["GDO"]
PARALLEL_UPDATES = 1 PARALLEL_UPDATES = 1
@ -32,118 +29,89 @@ async def async_setup_entry(
) -> None: ) -> None:
"""Set up Linear Garage Door cover.""" """Set up Linear Garage Door cover."""
coordinator: LinearUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id] coordinator: LinearUpdateCoordinator = hass.data[DOMAIN][config_entry.entry_id]
data = coordinator.data
device_list: list[LinearCoverEntity] = [] async_add_entities(
LinearCoverEntity(coordinator, device_id, sub_device_id)
for device_id in data: for device_id, device_data in coordinator.data.items()
device_list.extend( for sub_device_id in device_data.subdevices
LinearCoverEntity( if sub_device_id in SUPPORTED_SUBDEVICES
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)
class LinearCoverEntity(CoordinatorEntity[LinearUpdateCoordinator], CoverEntity): class LinearCoverEntity(CoordinatorEntity[LinearUpdateCoordinator], CoverEntity):
"""Representation of a Linear cover.""" """Representation of a Linear cover."""
_attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE _attr_supported_features = CoverEntityFeature.OPEN | CoverEntityFeature.CLOSE
_attr_has_entity_name = True
_attr_name = None
_attr_device_class = CoverDeviceClass.GARAGE
def __init__( def __init__(
self, self,
device_id: str,
device_name: str,
subdevice: str,
config_entry: ConfigEntry,
coordinator: LinearUpdateCoordinator, coordinator: LinearUpdateCoordinator,
device_id: str,
sub_device_id: str,
) -> None: ) -> None:
"""Init with device ID and name.""" """Init with device ID and name."""
super().__init__(coordinator) super().__init__(coordinator)
self._attr_has_entity_name = True
self._attr_name = None
self._device_id = device_id self._device_id = device_id
self._device_name = device_name self._sub_device_id = sub_device_id
self._subdevice = subdevice self._attr_unique_id = f"{device_id}-{sub_device_id}"
self._attr_device_class = CoverDeviceClass.GARAGE self._attr_device_info = DeviceInfo(
self._attr_unique_id = f"{device_id}-{subdevice}" identifiers={(DOMAIN, sub_device_id)},
self._config_entry = config_entry name=self.linear_device.name,
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,
manufacturer="Linear", manufacturer="Linear",
model="Garage Door Opener", 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 @property
def is_closed(self) -> bool: def is_closed(self) -> bool:
"""Return if cover is closed.""" """Return if cover is closed."""
return bool(self._get_data("Open_B") == "false") return self.sub_device.get("Open_B") == "false"
@property @property
def is_opened(self) -> bool: def is_opened(self) -> bool:
"""Return if cover is open.""" """Return if cover is open."""
return bool(self._get_data("Open_B") == "true") return self.sub_device.get("Open_B") == "true"
@property @property
def is_opening(self) -> bool: def is_opening(self) -> bool:
"""Return if cover is opening.""" """Return if cover is opening."""
return bool(self._get_data("Opening_P") == "0") return self.sub_device.get("Opening_P") == "0"
@property @property
def is_closing(self) -> bool: def is_closing(self) -> bool:
"""Return if cover is closing.""" """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: async def async_close_cover(self, **kwargs: Any) -> None:
"""Close the garage door.""" """Close the garage door."""
if self.is_closed: if self.is_closed:
return return
linear = Linear() await self.coordinator.execute(
lambda linear: linear.operate_device(
await linear.login( self._device_id, self._sub_device_id, "Close"
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 linear.operate_device(self._device_id, self._subdevice, "Close")
await linear.close()
async def async_open_cover(self, **kwargs: Any) -> None: async def async_open_cover(self, **kwargs: Any) -> None:
"""Open the garage door.""" """Open the garage door."""
if self.is_opened: if self.is_opened:
return return
linear = Linear() await self.coordinator.execute(
lambda linear: linear.operate_device(
await linear.login( self._device_id, self._sub_device_id, "Open"
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 linear.operate_device(self._device_id, self._subdevice, "Open")
await linear.close()

6
homeassistant/components/linear_garage_door/diagnostics.py
View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import asdict
from typing import Any from typing import Any
from homeassistant.components.diagnostics import async_redact_data from homeassistant.components.diagnostics import async_redact_data
@ -23,5 +24,8 @@ async def async_get_config_entry_diagnostics(
return { return {
"entry": async_redact_data(entry.as_dict(), TO_REDACT), "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()
},
} }

32
tests/components/linear_garage_door/test_cover.py
View File

@ -45,7 +45,7 @@ async def test_open_cover(hass: HomeAssistant) -> None:
await async_init_integration(hass) await async_init_integration(hass)
with patch( with patch(
"homeassistant.components.linear_garage_door.cover.Linear.operate_device" "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device"
) as operate_device: ) as operate_device:
await hass.services.async_call( await hass.services.async_call(
COVER_DOMAIN, COVER_DOMAIN,
@ -58,15 +58,15 @@ async def test_open_cover(hass: HomeAssistant) -> None:
with ( with (
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.login", "homeassistant.components.linear_garage_door.coordinator.Linear.login",
return_value=True, return_value=True,
), ),
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.operate_device", "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device",
return_value=None, return_value=None,
) as operate_device, ) as operate_device,
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.close", "homeassistant.components.linear_garage_door.coordinator.Linear.close",
return_value=True, return_value=True,
), ),
): ):
@ -80,11 +80,11 @@ async def test_open_cover(hass: HomeAssistant) -> None:
assert operate_device.call_count == 1 assert operate_device.call_count == 1
with ( with (
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.login", "homeassistant.components.linear_garage_door.coordinator.Linear.login",
return_value=True, return_value=True,
), ),
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.get_devices", "homeassistant.components.linear_garage_door.coordinator.Linear.get_devices",
return_value=[ return_value=[
{ {
"id": "test1", "id": "test1",
@ -99,7 +99,7 @@ async def test_open_cover(hass: HomeAssistant) -> None:
], ],
), ),
patch( 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: { side_effect=lambda id: {
"test1": { "test1": {
"GDO": {"Open_B": "true", "Open_P": "100"}, "GDO": {"Open_B": "true", "Open_P": "100"},
@ -120,7 +120,7 @@ async def test_open_cover(hass: HomeAssistant) -> None:
}[id], }[id],
), ),
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.close", "homeassistant.components.linear_garage_door.coordinator.Linear.close",
return_value=True, return_value=True,
), ),
): ):
@ -136,7 +136,7 @@ async def test_close_cover(hass: HomeAssistant) -> None:
await async_init_integration(hass) await async_init_integration(hass)
with patch( with patch(
"homeassistant.components.linear_garage_door.cover.Linear.operate_device" "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device"
) as operate_device: ) as operate_device:
await hass.services.async_call( await hass.services.async_call(
COVER_DOMAIN, COVER_DOMAIN,
@ -149,15 +149,15 @@ async def test_close_cover(hass: HomeAssistant) -> None:
with ( with (
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.login", "homeassistant.components.linear_garage_door.coordinator.Linear.login",
return_value=True, return_value=True,
), ),
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.operate_device", "homeassistant.components.linear_garage_door.coordinator.Linear.operate_device",
return_value=None, return_value=None,
) as operate_device, ) as operate_device,
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.close", "homeassistant.components.linear_garage_door.coordinator.Linear.close",
return_value=True, return_value=True,
), ),
): ):
@ -171,11 +171,11 @@ async def test_close_cover(hass: HomeAssistant) -> None:
assert operate_device.call_count == 1 assert operate_device.call_count == 1
with ( with (
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.login", "homeassistant.components.linear_garage_door.coordinator.Linear.login",
return_value=True, return_value=True,
), ),
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.get_devices", "homeassistant.components.linear_garage_door.coordinator.Linear.get_devices",
return_value=[ return_value=[
{ {
"id": "test1", "id": "test1",
@ -190,7 +190,7 @@ async def test_close_cover(hass: HomeAssistant) -> None:
], ],
), ),
patch( 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: { side_effect=lambda id: {
"test1": { "test1": {
"GDO": {"Open_B": "true", "Opening_P": "100"}, "GDO": {"Open_B": "true", "Opening_P": "100"},
@ -211,7 +211,7 @@ async def test_close_cover(hass: HomeAssistant) -> None:
}[id], }[id],
), ),
patch( patch(
"homeassistant.components.linear_garage_door.cover.Linear.close", "homeassistant.components.linear_garage_door.coordinator.Linear.close",
return_value=True, return_value=True,
), ),
): ):