Move venstar coordinator to separate module (#117500)

This commit is contained in:
epenet 2024-05-15 15:18:26 +02:00 committed by GitHub
parent d5a1587b1c
commit 4e600b7b19
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 85 additions and 74 deletions

View File

@ -1567,9 +1567,8 @@ omit =
homeassistant/components/velux/__init__.py
homeassistant/components/velux/cover.py
homeassistant/components/velux/light.py
homeassistant/components/venstar/__init__.py
homeassistant/components/venstar/binary_sensor.py
homeassistant/components/venstar/climate.py
homeassistant/components/venstar/coordinator.py
homeassistant/components/venstar/sensor.py
homeassistant/components/verisure/__init__.py
homeassistant/components/verisure/alarm_control_panel.py

View File

@ -2,10 +2,6 @@
from __future__ import annotations
import asyncio
from datetime import timedelta
from requests import RequestException
from venstarcolortouch import VenstarColorTouch
from homeassistant.config_entries import ConfigEntry
@ -18,11 +14,11 @@ from homeassistant.const import (
Platform,
)
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers import update_coordinator
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import _LOGGER, DOMAIN, VENSTAR_SLEEP, VENSTAR_TIMEOUT
from .const import DOMAIN, VENSTAR_TIMEOUT
from .coordinator import VenstarDataUpdateCoordinator
PLATFORMS = [Platform.BINARY_SENSOR, Platform.CLIMATE, Platform.SENSOR]
@ -65,67 +61,6 @@ async def async_unload_entry(hass: HomeAssistant, config: ConfigEntry) -> bool:
return unload_ok
class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator[None]): # pylint: disable=hass-enforce-coordinator-module
"""Class to manage fetching Venstar data."""
def __init__(
self,
hass: HomeAssistant,
*,
venstar_connection: VenstarColorTouch,
) -> None:
"""Initialize global Venstar data updater."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=60),
)
self.client = venstar_connection
self.runtimes: list[dict[str, int]] = []
async def _async_update_data(self) -> None:
"""Update the state."""
try:
await self.hass.async_add_executor_job(self.client.update_info)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar info update: {ex}"
) from ex
# older venstars sometimes cannot handle rapid sequential connections
await asyncio.sleep(VENSTAR_SLEEP)
try:
await self.hass.async_add_executor_job(self.client.update_sensors)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar sensor update: {ex}"
) from ex
# older venstars sometimes cannot handle rapid sequential connections
await asyncio.sleep(VENSTAR_SLEEP)
try:
await self.hass.async_add_executor_job(self.client.update_alerts)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar alert update: {ex}"
) from ex
# older venstars sometimes cannot handle rapid sequential connections
await asyncio.sleep(VENSTAR_SLEEP)
try:
self.runtimes = await self.hass.async_add_executor_job(
self.client.get_runtimes
)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar runtime update: {ex}"
) from ex
class VenstarEntity(CoordinatorEntity[VenstarDataUpdateCoordinator]):
"""Representation of a Venstar entity."""

View File

@ -36,7 +36,7 @@ import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from . import VenstarDataUpdateCoordinator, VenstarEntity
from . import VenstarEntity
from .const import (
_LOGGER,
ATTR_FAN_STATE,
@ -46,6 +46,7 @@ from .const import (
DOMAIN,
HOLD_MODE_TEMPERATURE,
)
from .coordinator import VenstarDataUpdateCoordinator
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{

View File

@ -0,0 +1,75 @@
"""Coordinator for the venstar component."""
from __future__ import annotations
import asyncio
from datetime import timedelta
from requests import RequestException
from venstarcolortouch import VenstarColorTouch
from homeassistant.core import HomeAssistant
from homeassistant.helpers import update_coordinator
from .const import _LOGGER, DOMAIN, VENSTAR_SLEEP
class VenstarDataUpdateCoordinator(update_coordinator.DataUpdateCoordinator[None]):
"""Class to manage fetching Venstar data."""
def __init__(
self,
hass: HomeAssistant,
*,
venstar_connection: VenstarColorTouch,
) -> None:
"""Initialize global Venstar data updater."""
super().__init__(
hass,
_LOGGER,
name=DOMAIN,
update_interval=timedelta(seconds=60),
)
self.client = venstar_connection
self.runtimes: list[dict[str, int]] = []
async def _async_update_data(self) -> None:
"""Update the state."""
try:
await self.hass.async_add_executor_job(self.client.update_info)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar info update: {ex}"
) from ex
# older venstars sometimes cannot handle rapid sequential connections
await asyncio.sleep(VENSTAR_SLEEP)
try:
await self.hass.async_add_executor_job(self.client.update_sensors)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar sensor update: {ex}"
) from ex
# older venstars sometimes cannot handle rapid sequential connections
await asyncio.sleep(VENSTAR_SLEEP)
try:
await self.hass.async_add_executor_job(self.client.update_alerts)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar alert update: {ex}"
) from ex
# older venstars sometimes cannot handle rapid sequential connections
await asyncio.sleep(VENSTAR_SLEEP)
try:
self.runtimes = await self.hass.async_add_executor_job(
self.client.get_runtimes
)
except (OSError, RequestException) as ex:
raise update_coordinator.UpdateFailed(
f"Exception during Venstar runtime update: {ex}"
) from ex

View File

@ -23,8 +23,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity import Entity
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from . import VenstarDataUpdateCoordinator, VenstarEntity
from . import VenstarEntity
from .const import DOMAIN
from .coordinator import VenstarDataUpdateCoordinator
RUNTIME_HEAT1 = "heat1"
RUNTIME_HEAT2 = "heat2"

View File

@ -20,7 +20,7 @@ EXPECTED_BASE_SUPPORTED_FEATURES = (
async def test_colortouch(hass: HomeAssistant) -> None:
"""Test interfacing with a venstar colortouch with attached humidifier."""
with patch("homeassistant.components.venstar.VENSTAR_SLEEP", new=0):
with patch("homeassistant.components.venstar.coordinator.VENSTAR_SLEEP", new=0):
await async_init_integration(hass)
state = hass.states.get("climate.colortouch")
@ -56,7 +56,7 @@ async def test_colortouch(hass: HomeAssistant) -> None:
async def test_t2000(hass: HomeAssistant) -> None:
"""Test interfacing with a venstar T2000 presently turned off."""
with patch("homeassistant.components.venstar.VENSTAR_SLEEP", new=0):
with patch("homeassistant.components.venstar.coordinator.VENSTAR_SLEEP", new=0):
await async_init_integration(hass)
state = hass.states.get("climate.t2000")

View File

@ -47,7 +47,7 @@ async def test_setup_entry(hass: HomeAssistant) -> None:
new=VenstarColorTouchMock.get_runtimes,
),
patch(
"homeassistant.components.venstar.VENSTAR_SLEEP",
"homeassistant.components.venstar.coordinator.VENSTAR_SLEEP",
new=0,
),
):