Move hunterdouglas_powerview data class to upstream library (#156228)

Co-authored-by: J. Nick Koston <nick@koston.org>

homeassistant/components/hunterdouglas_powerview/__init__.py

@@ -4,6 +4,7 @@ import logging
 from typing import TYPE_CHECKING
 
 from aiopvapi.resources.model import PowerviewData
+from aiopvapi.resources.shade_data import PowerviewShadeData
 from aiopvapi.rooms import Rooms
 from aiopvapi.scenes import Scenes
 from aiopvapi.shades import Shades
@@ -16,7 +17,6 @@ from homeassistant.helpers import device_registry as dr, entity_registry as er
 from .const import DOMAIN, HUB_EXCEPTIONS, MANUFACTURER
 from .coordinator import PowerviewShadeUpdateCoordinator
 from .model import PowerviewConfigEntry, PowerviewEntryData
-from .shade_data import PowerviewShadeData
 from .util import async_connect_hub
 
 PARALLEL_UPDATES = 1

homeassistant/components/hunterdouglas_powerview/coordinator.py

@@ -8,6 +8,7 @@ import logging
 
 from aiopvapi.helpers.aiorequest import PvApiMaintenance
 from aiopvapi.hub import Hub
+from aiopvapi.resources.shade_data import PowerviewShadeData
 from aiopvapi.shades import Shades
 
 from homeassistant.config_entries import ConfigEntry
@@ -15,7 +16,6 @@ from homeassistant.core import HomeAssistant
 from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import HUB_EXCEPTIONS
-from .shade_data import PowerviewShadeData
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/hunterdouglas_powerview/cover.py

@@ -208,13 +208,13 @@ class PowerViewShadeBase(ShadeEntity, CoverEntity):
     async def _async_execute_move(self, move: ShadePosition) -> None:
         """Execute a move that can affect multiple positions."""
         _LOGGER.debug("Move request %s: %s", self.name, move)
+        # Store the requested positions so subsequent move
+        # requests contain the secondary shade positions
+        self.data.update_shade_position(self._shade.id, move)
         async with self.coordinator.radio_operation_lock:
             response = await self._shade.move(move)
         _LOGGER.debug("Move response %s: %s", self.name, response)
 
-        # Process the response from the hub (including new positions)
-        self.data.update_shade_position(self._shade.id, response)
-
     async def _async_set_cover_position(self, target_hass_position: int) -> None:
         """Move the shade to a position."""
         target_hass_position = self._clamp_cover_limit(target_hass_position)

homeassistant/components/hunterdouglas_powerview/entity.py

@@ -3,6 +3,7 @@
 import logging
 
 from aiopvapi.resources.shade import BaseShade, ShadePosition
+from aiopvapi.resources.shade_data import PowerviewShadeData
 
 from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers.device_registry import DeviceInfo
@@ -11,7 +12,6 @@ from homeassistant.helpers.update_coordinator import CoordinatorEntity
 from .const import DOMAIN, MANUFACTURER
 from .coordinator import PowerviewShadeUpdateCoordinator
 from .model import PowerviewDeviceInfo
-from .shade_data import PowerviewShadeData
 
 _LOGGER = logging.getLogger(__name__)
 

homeassistant/components/hunterdouglas_powerview/shade_data.py

@@ -1,80 +0,0 @@
-"""Shade data for the Hunter Douglas PowerView integration."""
-
-from __future__ import annotations
-
-from dataclasses import fields
-from typing import Any
-
-from aiopvapi.resources.model import PowerviewData
-from aiopvapi.resources.shade import BaseShade, ShadePosition
-
-from .util import async_map_data_by_id
-
-POSITION_FIELDS = [field for field in fields(ShadePosition) if field.name != "velocity"]
-
-
-def copy_position_data(source: ShadePosition, target: ShadePosition) -> ShadePosition:
-    """Copy position data from source to target for None values only."""
-    for field in POSITION_FIELDS:
-        if (value := getattr(source, field.name)) is not None:
-            setattr(target, field.name, value)
-
-
-class PowerviewShadeData:
-    """Coordinate shade data between multiple api calls."""
-
-    def __init__(self) -> None:
-        """Init the shade data."""
-        self._raw_data_by_id: dict[int, dict[str | int, Any]] = {}
-        self._shade_group_data_by_id: dict[int, BaseShade] = {}
-        self.positions: dict[int, ShadePosition] = {}
-
-    def get_raw_data(self, shade_id: int) -> dict[str | int, Any]:
-        """Get data for the shade."""
-        return self._raw_data_by_id[shade_id]
-
-    def get_all_raw_data(self) -> dict[int, dict[str | int, Any]]:
-        """Get data for all shades."""
-        return self._raw_data_by_id
-
-    def get_shade(self, shade_id: int) -> BaseShade:
-        """Get specific shade from the coordinator."""
-        return self._shade_group_data_by_id[shade_id]
-
-    def get_shade_position(self, shade_id: int) -> ShadePosition:
-        """Get positions for a shade."""
-        if shade_id not in self.positions:
-            shade_position = ShadePosition()
-            # If we have the group data, use it to populate the initial position
-            if shade := self._shade_group_data_by_id.get(shade_id):
-                copy_position_data(shade.current_position, shade_position)
-            self.positions[shade_id] = shade_position
-        return self.positions[shade_id]
-
-    def update_from_group_data(self, shade_id: int) -> None:
-        """Process an update from the group data."""
-        data = self._shade_group_data_by_id[shade_id]
-        copy_position_data(data.current_position, self.get_shade_position(data.id))
-
-    def store_group_data(self, shade_data: PowerviewData) -> None:
-        """Store data from the all shades endpoint.
-
-        This does not update the shades or positions (self.positions)
-        as the data may be stale. update_from_group_data
-        with a shade_id will update a specific shade
-        from the group data.
-        """
-        self._shade_group_data_by_id = shade_data.processed
-        self._raw_data_by_id = async_map_data_by_id(shade_data.raw)
-
-    def update_shade_position(self, shade_id: int, new_position: ShadePosition) -> None:
-        """Update a single shades position."""
-        copy_position_data(new_position, self.get_shade_position(shade_id))
-
-    def update_shade_velocity(self, shade_id: int, shade_data: ShadePosition) -> None:
-        """Update a single shades velocity."""
-        # the hub will always return a velocity of 0 on initial connect,
-        # separate definition to store consistent value in HA
-        # this value is purely driven from HA
-        if shade_data.velocity is not None:
-            self.get_shade_position(shade_id).velocity = shade_data.velocity

homeassistant/components/hunterdouglas_powerview/util.py

@@ -2,25 +2,15 @@
 
 from __future__ import annotations
 
-from collections.abc import Iterable
-from typing import Any
-
 from aiopvapi.helpers.aiorequest import AioRequest
-from aiopvapi.helpers.constants import ATTR_ID
 from aiopvapi.hub import Hub
 
-from homeassistant.core import HomeAssistant, callback
+from homeassistant.core import HomeAssistant
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
 
 from .model import PowerviewAPI, PowerviewDeviceInfo
 
 
-@callback
-def async_map_data_by_id(data: Iterable[dict[str | int, Any]]):
-    """Return a dict with the key being the id for a list of entries."""
-    return {entry[ATTR_ID]: entry for entry in data}
-
-
 async def async_connect_hub(
     hass: HomeAssistant, address: str, api_version: int | None = None
 ) -> PowerviewAPI: