diff --git a/homeassistant/components/sms/__init__.py b/homeassistant/components/sms/__init__.py index 824a95e36b1..a606b83896f 100644 --- a/homeassistant/components/sms/__init__.py +++ b/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 diff --git a/homeassistant/components/sms/coordinator.py b/homeassistant/components/sms/coordinator.py new file mode 100644 index 00000000000..fd212fce4f2 --- /dev/null +++ b/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