Move gogogate2 coordinator to separate module (#117433)

This commit is contained in:
epenet 2024-05-16 09:11:49 +02:00 committed by GitHub
parent 07d289d1c6
commit 8bbac8040f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 52 additions and 51 deletions

View File

@ -2,7 +2,7 @@
from __future__ import annotations
from collections.abc import Awaitable, Callable, Mapping
from collections.abc import Mapping
from datetime import timedelta
import logging
from typing import Any, NamedTuple
@ -24,16 +24,12 @@ from homeassistant.const import (
CONF_USERNAME,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.httpx_client import get_async_client
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
UpdateFailed,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity, UpdateFailed
from .const import DATA_UPDATE_COORDINATOR, DEVICE_TYPE_ISMARTGATE, DOMAIN, MANUFACTURER
from .coordinator import DeviceDataUpdateCoordinator
_LOGGER = logging.getLogger(__name__)
@ -46,38 +42,6 @@ class StateData(NamedTuple):
door: AbstractDoor | None
class DeviceDataUpdateCoordinator(
DataUpdateCoordinator[GogoGate2InfoResponse | ISmartGateInfoResponse]
): # pylint: disable=hass-enforce-coordinator-module
"""Manages polling for state changes from the device."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
api: AbstractGateApi,
*,
name: str,
update_interval: timedelta,
update_method: Callable[
[], Awaitable[GogoGate2InfoResponse | ISmartGateInfoResponse]
]
| None = None,
request_refresh_debouncer: Debouncer | None = None,
) -> None:
"""Initialize the data update coordinator."""
DataUpdateCoordinator.__init__(
self,
hass,
logger,
name=name,
update_interval=update_interval,
update_method=update_method,
request_refresh_debouncer=request_refresh_debouncer,
)
self.api = api
class GoGoGate2Entity(CoordinatorEntity[DeviceDataUpdateCoordinator]):
"""Base class for gogogate2 entities."""

View File

@ -0,0 +1,45 @@
"""Coordinator for GogoGate2 component."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from datetime import timedelta
import logging
from ismartgate import AbstractGateApi, GogoGate2InfoResponse, ISmartGateInfoResponse
from homeassistant.core import HomeAssistant
from homeassistant.helpers.debounce import Debouncer
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
class DeviceDataUpdateCoordinator(
DataUpdateCoordinator[GogoGate2InfoResponse | ISmartGateInfoResponse]
):
"""Manages polling for state changes from the device."""
def __init__(
self,
hass: HomeAssistant,
logger: logging.Logger,
api: AbstractGateApi,
*,
name: str,
update_interval: timedelta,
update_method: Callable[
[], Awaitable[GogoGate2InfoResponse | ISmartGateInfoResponse]
]
| None = None,
request_refresh_debouncer: Debouncer | None = None,
) -> None:
"""Initialize the data update coordinator."""
DataUpdateCoordinator.__init__(
self,
hass,
logger,
name=name,
update_interval=update_interval,
update_method=update_method,
request_refresh_debouncer=request_refresh_debouncer,
)
self.api = api

View File

@ -20,12 +20,8 @@ from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .common import (
DeviceDataUpdateCoordinator,
GoGoGate2Entity,
cover_unique_id,
get_data_update_coordinator,
)
from .common import GoGoGate2Entity, cover_unique_id, get_data_update_coordinator
from .coordinator import DeviceDataUpdateCoordinator
async def async_setup_entry(

View File

@ -16,12 +16,8 @@ from homeassistant.const import PERCENTAGE, EntityCategory, UnitOfTemperature
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from .common import (
DeviceDataUpdateCoordinator,
GoGoGate2Entity,
get_data_update_coordinator,
sensor_unique_id,
)
from .common import GoGoGate2Entity, get_data_update_coordinator, sensor_unique_id
from .coordinator import DeviceDataUpdateCoordinator
SENSOR_ID_WIRED = "WIRE"