Explicitly pass in the config_entry in iron_os coordinator (#138137)

explicitly pass in the config_entry in coordinator
pull/138165/head
Michael 2025-02-09 21:42:15 +01:00 committed by GitHub
parent 5dea4164a5
commit 427013124c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 40 additions and 32 deletions

View File

@ -8,7 +8,6 @@ from typing import TYPE_CHECKING
from pynecil import IronOSUpdate, Pynecil
from homeassistant.components import bluetooth
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_NAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryNotReady
@ -19,6 +18,7 @@ from homeassistant.util.hass_dict import HassKey
from .const import DOMAIN
from .coordinator import (
IronOSConfigEntry,
IronOSCoordinators,
IronOSFirmwareUpdateCoordinator,
IronOSLiveDataCoordinator,
@ -39,7 +39,6 @@ PLATFORMS: list[Platform] = [
CONFIG_SCHEMA = cv.config_entry_only_config_schema(DOMAIN)
type IronOSConfigEntry = ConfigEntry[IronOSCoordinators]
IRON_OS_KEY: HassKey[IronOSFirmwareUpdateCoordinator] = HassKey(DOMAIN)
@ -73,10 +72,10 @@ async def async_setup_entry(hass: HomeAssistant, entry: IronOSConfigEntry) -> bo
device = Pynecil(ble_device)
live_data = IronOSLiveDataCoordinator(hass, device)
live_data = IronOSLiveDataCoordinator(hass, entry, device)
await live_data.async_config_entry_first_refresh()
settings = IronOSSettingsCoordinator(hass, device)
settings = IronOSSettingsCoordinator(hass, entry, device)
await settings.async_config_entry_first_refresh()
entry.runtime_data = IronOSCoordinators(

View File

@ -43,15 +43,19 @@ class IronOSCoordinators:
settings: IronOSSettingsCoordinator
type IronOSConfigEntry = ConfigEntry[IronOSCoordinators]
class IronOSBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
"""IronOS base coordinator."""
device_info: DeviceInfoResponse
config_entry: ConfigEntry
config_entry: IronOSConfigEntry
def __init__(
self,
hass: HomeAssistant,
config_entry: IronOSConfigEntry,
device: Pynecil,
update_interval: timedelta,
) -> None:
@ -60,6 +64,7 @@ class IronOSBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
super().__init__(
hass,
_LOGGER,
config_entry=config_entry,
name=DOMAIN,
update_interval=update_interval,
request_refresh_debouncer=Debouncer(
@ -80,9 +85,11 @@ class IronOSBaseCoordinator[_DataT](DataUpdateCoordinator[_DataT]):
class IronOSLiveDataCoordinator(IronOSBaseCoordinator[LiveDataResponse]):
"""IronOS coordinator."""
def __init__(self, hass: HomeAssistant, device: Pynecil) -> None:
def __init__(
self, hass: HomeAssistant, config_entry: IronOSConfigEntry, device: Pynecil
) -> None:
"""Initialize IronOS coordinator."""
super().__init__(hass, device=device, update_interval=SCAN_INTERVAL)
super().__init__(hass, config_entry, device, SCAN_INTERVAL)
async def _async_update_data(self) -> LiveDataResponse:
"""Fetch data from Device."""
@ -109,35 +116,14 @@ class IronOSLiveDataCoordinator(IronOSBaseCoordinator[LiveDataResponse]):
return False
class IronOSFirmwareUpdateCoordinator(DataUpdateCoordinator[LatestRelease]):
"""IronOS coordinator for retrieving update information from github."""
def __init__(self, hass: HomeAssistant, github: IronOSUpdate) -> None:
"""Initialize IronOS coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=None,
name=DOMAIN,
update_interval=SCAN_INTERVAL_GITHUB,
)
self.github = github
async def _async_update_data(self) -> LatestRelease:
"""Fetch data from Github."""
try:
return await self.github.latest_release()
except UpdateException as e:
raise UpdateFailed("Failed to check for latest IronOS update") from e
class IronOSSettingsCoordinator(IronOSBaseCoordinator[SettingsDataResponse]):
"""IronOS coordinator."""
def __init__(self, hass: HomeAssistant, device: Pynecil) -> None:
def __init__(
self, hass: HomeAssistant, config_entry: IronOSConfigEntry, device: Pynecil
) -> None:
"""Initialize IronOS coordinator."""
super().__init__(hass, device=device, update_interval=SCAN_INTERVAL_SETTINGS)
super().__init__(hass, config_entry, device, SCAN_INTERVAL_SETTINGS)
async def _async_update_data(self) -> SettingsDataResponse:
"""Fetch data from Device."""
@ -173,3 +159,26 @@ class IronOSSettingsCoordinator(IronOSBaseCoordinator[SettingsDataResponse]):
)
self.async_update_listeners()
await self.async_request_refresh()
class IronOSFirmwareUpdateCoordinator(DataUpdateCoordinator[LatestRelease]):
"""IronOS coordinator for retrieving update information from github."""
def __init__(self, hass: HomeAssistant, github: IronOSUpdate) -> None:
"""Initialize IronOS coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=None,
name=DOMAIN,
update_interval=SCAN_INTERVAL_GITHUB,
)
self.github = github
async def _async_update_data(self) -> LatestRelease:
"""Fetch data from Github."""
try:
return await self.github.latest_release()
except UpdateException as e:
raise UpdateFailed("Failed to check for latest IronOS update") from e