mirror of
https://github.com/home-assistant/core.git
synced 2025-07-23 21:27:38 +00:00
Move Airthings coordinator to separate module (#145827)
* Create coordinator * Fix sensor.py
This commit is contained in:
parent
afa97f8ec1
commit
e2fc2dce84
@ -5,23 +5,22 @@ from __future__ import annotations
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from airthings import Airthings, AirthingsDevice, AirthingsError
|
||||
from airthings import Airthings
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONF_ID, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import CONF_SECRET, DOMAIN
|
||||
from .const import CONF_SECRET
|
||||
from .coordinator import AirthingsDataUpdateCoordinator
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
SCAN_INTERVAL = timedelta(minutes=6)
|
||||
|
||||
type AirthingsDataCoordinatorType = DataUpdateCoordinator[dict[str, AirthingsDevice]]
|
||||
type AirthingsConfigEntry = ConfigEntry[AirthingsDataCoordinatorType]
|
||||
type AirthingsConfigEntry = ConfigEntry[AirthingsDataUpdateCoordinator]
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: AirthingsConfigEntry) -> bool:
|
||||
@ -32,21 +31,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: AirthingsConfigEntry) ->
|
||||
async_get_clientsession(hass),
|
||||
)
|
||||
|
||||
async def _update_method() -> dict[str, AirthingsDevice]:
|
||||
"""Get the latest data from Airthings."""
|
||||
try:
|
||||
return await airthings.update_devices() # type: ignore[no-any-return]
|
||||
except AirthingsError as err:
|
||||
raise UpdateFailed(f"Unable to fetch data: {err}") from err
|
||||
coordinator = AirthingsDataUpdateCoordinator(hass, airthings)
|
||||
|
||||
coordinator = DataUpdateCoordinator(
|
||||
hass,
|
||||
_LOGGER,
|
||||
config_entry=entry,
|
||||
name=DOMAIN,
|
||||
update_method=_update_method,
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
|
||||
entry.runtime_data = coordinator
|
||||
|
36
homeassistant/components/airthings/coordinator.py
Normal file
36
homeassistant/components/airthings/coordinator.py
Normal file
@ -0,0 +1,36 @@
|
||||
"""The Airthings integration."""
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
|
||||
from airthings import Airthings, AirthingsDevice, AirthingsError
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
SCAN_INTERVAL = timedelta(minutes=6)
|
||||
|
||||
|
||||
class AirthingsDataUpdateCoordinator(DataUpdateCoordinator[dict[str, AirthingsDevice]]):
|
||||
"""Coordinator for Airthings data updates."""
|
||||
|
||||
def __init__(self, hass: HomeAssistant, airthings: Airthings) -> None:
|
||||
"""Initialize the coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=DOMAIN,
|
||||
update_method=self._update_method,
|
||||
update_interval=SCAN_INTERVAL,
|
||||
)
|
||||
self.airthings = airthings
|
||||
|
||||
async def _update_method(self) -> dict[str, AirthingsDevice]:
|
||||
"""Get the latest data from Airthings."""
|
||||
try:
|
||||
return await self.airthings.update_devices() # type: ignore[no-any-return]
|
||||
except AirthingsError as err:
|
||||
raise UpdateFailed(f"Unable to fetch data: {err}") from err
|
@ -27,8 +27,9 @@ from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
from homeassistant.helpers.typing import StateType
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from . import AirthingsConfigEntry, AirthingsDataCoordinatorType
|
||||
from . import AirthingsConfigEntry
|
||||
from .const import DOMAIN
|
||||
from .coordinator import AirthingsDataUpdateCoordinator
|
||||
|
||||
SENSORS: dict[str, SensorEntityDescription] = {
|
||||
"radonShortTermAvg": SensorEntityDescription(
|
||||
@ -140,7 +141,7 @@ async def async_setup_entry(
|
||||
|
||||
|
||||
class AirthingsHeaterEnergySensor(
|
||||
CoordinatorEntity[AirthingsDataCoordinatorType], SensorEntity
|
||||
CoordinatorEntity[AirthingsDataUpdateCoordinator], SensorEntity
|
||||
):
|
||||
"""Representation of a Airthings Sensor device."""
|
||||
|
||||
@ -149,7 +150,7 @@ class AirthingsHeaterEnergySensor(
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: AirthingsDataCoordinatorType,
|
||||
coordinator: AirthingsDataUpdateCoordinator,
|
||||
airthings_device: AirthingsDevice,
|
||||
entity_description: SensorEntityDescription,
|
||||
) -> None:
|
||||
|
Loading…
x
Reference in New Issue
Block a user