Move motion blinds coordinator to its own file (#100952)

.coveragerc

@@ -767,6 +767,7 @@ omit =
     homeassistant/components/moehlenhoff_alpha2/climate.py
     homeassistant/components/moehlenhoff_alpha2/sensor.py
     homeassistant/components/motion_blinds/__init__.py
+    homeassistant/components/motion_blinds/coordinator.py
     homeassistant/components/motion_blinds/cover.py
     homeassistant/components/motion_blinds/entity.py
     homeassistant/components/motion_blinds/sensor.py

homeassistant/components/motion_blinds/__init__.py

@@ -2,19 +2,16 @@
 import asyncio
 from datetime import timedelta
 import logging
-from socket import timeout
+from typing import TYPE_CHECKING
-from typing import TYPE_CHECKING, Any
 
-from motionblinds import AsyncMotionMulticast, ParseException
+from motionblinds import AsyncMotionMulticast
 
 from homeassistant.config_entries import ConfigEntry, ConfigEntryState
 from homeassistant.const import CONF_API_KEY, CONF_HOST, EVENT_HOMEASSISTANT_STOP
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
 
 from .const import (
-    ATTR_AVAILABLE,
     CONF_INTERFACE,
     CONF_WAIT_FOR_PUSH,
     DEFAULT_INTERFACE,
@@ -28,85 +25,13 @@ from .const import (
     KEY_UNSUB_STOP,
     PLATFORMS,
     UPDATE_INTERVAL,
-    UPDATE_INTERVAL_FAST,
 )
+from .coordinator import DataUpdateCoordinatorMotionBlinds
 from .gateway import ConnectMotionGateway
 
 _LOGGER = logging.getLogger(__name__)
 
 
-class DataUpdateCoordinatorMotionBlinds(DataUpdateCoordinator):
-    """Class to manage fetching data from single endpoint."""
-
-    def __init__(
-        self,
-        hass: HomeAssistant,
-        logger: logging.Logger,
-        coordinator_info: dict[str, Any],
-        *,
-        name: str,
-        update_interval: timedelta,
-    ) -> None:
-        """Initialize global data updater."""
-        super().__init__(
-            hass,
-            logger,
-            name=name,
-            update_interval=update_interval,
-        )
-
-        self.api_lock = coordinator_info[KEY_API_LOCK]
-        self._gateway = coordinator_info[KEY_GATEWAY]
-        self._wait_for_push = coordinator_info[CONF_WAIT_FOR_PUSH]
-
-    def update_gateway(self):
-        """Fetch data from gateway."""
-        try:
-            self._gateway.Update()
-        except (timeout, ParseException):
-            # let the error be logged and handled by the motionblinds library
-            return {ATTR_AVAILABLE: False}
-
-        return {ATTR_AVAILABLE: True}
-
-    def update_blind(self, blind):
-        """Fetch data from a blind."""
-        try:
-            if self._wait_for_push:
-                blind.Update()
-            else:
-                blind.Update_trigger()
-        except (timeout, ParseException):
-            # let the error be logged and handled by the motionblinds library
-            return {ATTR_AVAILABLE: False}
-
-        return {ATTR_AVAILABLE: True}
-
-    async def _async_update_data(self):
-        """Fetch the latest data from the gateway and blinds."""
-        data = {}
-
-        async with self.api_lock:
-            data[KEY_GATEWAY] = await self.hass.async_add_executor_job(
-                self.update_gateway
-            )
-
-        for blind in self._gateway.device_list.values():
-            await asyncio.sleep(1.5)
-            async with self.api_lock:
-                data[blind.mac] = await self.hass.async_add_executor_job(
-                    self.update_blind, blind
-                )
-
-        all_available = all(device[ATTR_AVAILABLE] for device in data.values())
-        if all_available:
-            self.update_interval = timedelta(seconds=UPDATE_INTERVAL)
-        else:
-            self.update_interval = timedelta(seconds=UPDATE_INTERVAL_FAST)
-
-        return data
-
-
 async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     """Set up the motion_blinds components from a config entry."""
     hass.data.setdefault(DOMAIN, {})

homeassistant/components/motion_blinds/coordinator.py

@@ -0,0 +1,94 @@
+"""DataUpdateCoordinator for motion blinds integration."""
+import asyncio
+from datetime import timedelta
+import logging
+from socket import timeout
+from typing import Any
+
+from motionblinds import ParseException
+
+from homeassistant.core import HomeAssistant
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
+
+from .const import (
+    ATTR_AVAILABLE,
+    CONF_WAIT_FOR_PUSH,
+    KEY_API_LOCK,
+    KEY_GATEWAY,
+    UPDATE_INTERVAL,
+    UPDATE_INTERVAL_FAST,
+)
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class DataUpdateCoordinatorMotionBlinds(DataUpdateCoordinator):
+    """Class to manage fetching data from single endpoint."""
+
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        logger: logging.Logger,
+        coordinator_info: dict[str, Any],
+        *,
+        name: str,
+        update_interval: timedelta,
+    ) -> None:
+        """Initialize global data updater."""
+        super().__init__(
+            hass,
+            logger,
+            name=name,
+            update_interval=update_interval,
+        )
+
+        self.api_lock = coordinator_info[KEY_API_LOCK]
+        self._gateway = coordinator_info[KEY_GATEWAY]
+        self._wait_for_push = coordinator_info[CONF_WAIT_FOR_PUSH]
+
+    def update_gateway(self):
+        """Fetch data from gateway."""
+        try:
+            self._gateway.Update()
+        except (timeout, ParseException):
+            # let the error be logged and handled by the motionblinds library
+            return {ATTR_AVAILABLE: False}
+
+        return {ATTR_AVAILABLE: True}
+
+    def update_blind(self, blind):
+        """Fetch data from a blind."""
+        try:
+            if self._wait_for_push:
+                blind.Update()
+            else:
+                blind.Update_trigger()
+        except (timeout, ParseException):
+            # let the error be logged and handled by the motionblinds library
+            return {ATTR_AVAILABLE: False}
+
+        return {ATTR_AVAILABLE: True}
+
+    async def _async_update_data(self):
+        """Fetch the latest data from the gateway and blinds."""
+        data = {}
+
+        async with self.api_lock:
+            data[KEY_GATEWAY] = await self.hass.async_add_executor_job(
+                self.update_gateway
+            )
+
+        for blind in self._gateway.device_list.values():
+            await asyncio.sleep(1.5)
+            async with self.api_lock:
+                data[blind.mac] = await self.hass.async_add_executor_job(
+                    self.update_blind, blind
+                )
+
+        all_available = all(device[ATTR_AVAILABLE] for device in data.values())
+        if all_available:
+            self.update_interval = timedelta(seconds=UPDATE_INTERVAL)
+        else:
+            self.update_interval = timedelta(seconds=UPDATE_INTERVAL_FAST)
+
+        return data

homeassistant/components/motion_blinds/entity.py

@@ -8,7 +8,6 @@ from homeassistant.helpers import device_registry as dr
 from homeassistant.helpers.device_registry import DeviceInfo
 from homeassistant.helpers.update_coordinator import CoordinatorEntity
 
-from . import DataUpdateCoordinatorMotionBlinds
 from .const import (
     ATTR_AVAILABLE,
     DEFAULT_GATEWAY_NAME,
@@ -16,6 +15,7 @@ from .const import (
     KEY_GATEWAY,
     MANUFACTURER,
 )
+from .coordinator import DataUpdateCoordinatorMotionBlinds
 from .gateway import device_name