Move sms coordinators to their own file (#100276)

homeassistant/components/sms/__init__.py

@@ -1,9 +1,6 @@
 """The sms component."""
-import asyncio
-from datetime import timedelta
 import logging
 
-import gammu
 import voluptuous as vol
 
 from homeassistant.config_entries import ConfigEntry
@@ -12,12 +9,10 @@ from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import config_validation as cv, discovery
 from homeassistant.helpers.typing import ConfigType
-from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
 
 from .const import (
     CONF_BAUD_SPEED,
     DEFAULT_BAUD_SPEED,
-    DEFAULT_SCAN_INTERVAL,
     DOMAIN,
     GATEWAY,
     HASS_CONFIG,
@@ -25,6 +20,7 @@ from .const import (
     SIGNAL_COORDINATOR,
     SMS_GATEWAY,
 )
+from .coordinator import NetworkCoordinator, SignalCoordinator
 from .gateway import create_sms_gateway
 
 _LOGGER = logging.getLogger(__name__)
@@ -45,8 +41,6 @@ CONFIG_SCHEMA = vol.Schema(
     extra=vol.ALLOW_EXTRA,
 )
 
-_LOGGER = logging.getLogger(__name__)
-
 
 async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
     """Configure Gammu state machine."""
@@ -107,47 +101,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
         await gateway.terminate_async()
 
     return unload_ok
-
-
-class SignalCoordinator(DataUpdateCoordinator):
-    """Signal strength coordinator."""
-
-    def __init__(self, hass, gateway):
-        """Initialize signal strength coordinator."""
-        super().__init__(
-            hass,
-            _LOGGER,
-            name="Device signal state",
-            update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
-        )
-        self._gateway = gateway
-
-    async def _async_update_data(self):
-        """Fetch device signal quality."""
-        try:
-            async with asyncio.timeout(10):
-                return await self._gateway.get_signal_quality_async()
-        except gammu.GSMError as exc:
-            raise UpdateFailed(f"Error communicating with device: {exc}") from exc
-
-
-class NetworkCoordinator(DataUpdateCoordinator):
-    """Network info coordinator."""
-
-    def __init__(self, hass, gateway):
-        """Initialize network info coordinator."""
-        super().__init__(
-            hass,
-            _LOGGER,
-            name="Device network state",
-            update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
-        )
-        self._gateway = gateway
-
-    async def _async_update_data(self):
-        """Fetch device network info."""
-        try:
-            async with asyncio.timeout(10):
-                return await self._gateway.get_network_info_async()
-        except gammu.GSMError as exc:
-            raise UpdateFailed(f"Error communicating with device: {exc}") from exc

homeassistant/components/sms/coordinator.py

@@ -0,0 +1,56 @@
+"""DataUpdateCoordinators for the sms integration."""
+import asyncio
+from datetime import timedelta
+import logging
+
+import gammu
+
+from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
+
+from .const import DEFAULT_SCAN_INTERVAL
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class SignalCoordinator(DataUpdateCoordinator):
+    """Signal strength coordinator."""
+
+    def __init__(self, hass, gateway):
+        """Initialize signal strength coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Device signal state",
+            update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
+        )
+        self._gateway = gateway
+
+    async def _async_update_data(self):
+        """Fetch device signal quality."""
+        try:
+            async with asyncio.timeout(10):
+                return await self._gateway.get_signal_quality_async()
+        except gammu.GSMError as exc:
+            raise UpdateFailed(f"Error communicating with device: {exc}") from exc
+
+
+class NetworkCoordinator(DataUpdateCoordinator):
+    """Network info coordinator."""
+
+    def __init__(self, hass, gateway):
+        """Initialize network info coordinator."""
+        super().__init__(
+            hass,
+            _LOGGER,
+            name="Device network state",
+            update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
+        )
+        self._gateway = gateway
+
+    async def _async_update_data(self):
+        """Fetch device network info."""
+        try:
+            async with asyncio.timeout(10):
+                return await self._gateway.get_network_info_async()
+        except gammu.GSMError as exc:
+            raise UpdateFailed(f"Error communicating with device: {exc}") from exc