Move ialarm coordinator to separate module (#117478)

This commit is contained in:
epenet 2024-05-15 09:30:44 +02:00 committed by GitHub
parent 72d873ce70
commit a36ad6bb64
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 52 additions and 39 deletions

View file

@ -3,21 +3,18 @@
from __future__ import annotations
import asyncio
import logging
from pyialarm import IAlarm
from homeassistant.components.alarm_control_panel import SCAN_INTERVAL
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_HOST, CONF_PORT, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DATA_COORDINATOR, DOMAIN, IALARM_TO_HASS
from .const import DATA_COORDINATOR, DOMAIN
from .coordinator import IAlarmDataUpdateCoordinator
PLATFORMS = [Platform.ALARM_CONTROL_PANEL]
_LOGGER = logging.getLogger(__name__)
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
@ -52,36 +49,3 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
class IAlarmDataUpdateCoordinator(DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching iAlarm data."""
def __init__(self, hass: HomeAssistant, ialarm: IAlarm, mac: str) -> None:
"""Initialize global iAlarm data updater."""
self.ialarm = ialarm
self.state: str | None = None
self.host: str = ialarm.host
self.mac = mac
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
)
def _update_data(self) -> None:
"""Fetch data from iAlarm via sync functions."""
status = self.ialarm.get_status()
_LOGGER.debug("iAlarm status: %s", status)
self.state = IALARM_TO_HASS.get(status)
async def _async_update_data(self) -> None:
"""Fetch data from iAlarm."""
try:
async with asyncio.timeout(10):
await self.hass.async_add_executor_job(self._update_data)
except ConnectionError as error:
raise UpdateFailed(error) from error

View file

@ -12,8 +12,8 @@ from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from . import IAlarmDataUpdateCoordinator
from .const import DATA_COORDINATOR, DOMAIN
from .coordinator import IAlarmDataUpdateCoordinator
async def async_setup_entry(

View file

@ -0,0 +1,49 @@
"""Coordinator for the iAlarm integration."""
from __future__ import annotations
import asyncio
import logging
from pyialarm import IAlarm
from homeassistant.components.alarm_control_panel import SCAN_INTERVAL
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from .const import DOMAIN, IALARM_TO_HASS
_LOGGER = logging.getLogger(__name__)
class IAlarmDataUpdateCoordinator(DataUpdateCoordinator[None]):
"""Class to manage fetching iAlarm data."""
def __init__(self, hass: HomeAssistant, ialarm: IAlarm, mac: str) -> None:
"""Initialize global iAlarm data updater."""
self.ialarm = ialarm
self.state: str | None = None
self.host: str = ialarm.host
self.mac = mac
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=SCAN_INTERVAL,
)
def _update_data(self) -> None:
"""Fetch data from iAlarm via sync functions."""
status = self.ialarm.get_status()
_LOGGER.debug("iAlarm status: %s", status)
self.state = IALARM_TO_HASS.get(status)
async def _async_update_data(self) -> None:
"""Fetch data from iAlarm."""
try:
async with asyncio.timeout(10):
await self.hass.async_add_executor_job(self._update_data)
except ConnectionError as error:
raise UpdateFailed(error) from error