Use config entry runtime_data in aussie broadband (#127083)

This commit is contained in:
epenet 2024-09-30 11:04:08 +02:00 committed by GitHub
parent 40f808e9be
commit c3c2bc51c5
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 49 additions and 29 deletions

View File

@ -7,19 +7,22 @@ from aussiebb.asyncio import AussieBB
from aussiebb.const import FETCH_TYPES
from aussiebb.exceptions import AuthenticationException
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_PASSWORD, CONF_USERNAME, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryAuthFailed, ConfigEntryNotReady
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import DOMAIN
from .coordinator import AussieBroadandDataUpdateCoordinator
from .coordinator import (
AussieBroadbandConfigEntry,
AussieBroadbandDataUpdateCoordinator,
)
PLATFORMS = [Platform.SENSOR]
async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_setup_entry(
hass: HomeAssistant, entry: AussieBroadbandConfigEntry
) -> bool:
"""Set up Aussie Broadband from a config entry."""
# Login to the Aussie Broadband API and retrieve the current service list
client = AussieBB(
@ -41,25 +44,20 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
# Initiate a Data Update Coordinator for each service
for service in services:
service["coordinator"] = AussieBroadandDataUpdateCoordinator(
service["coordinator"] = AussieBroadbandDataUpdateCoordinator(
hass, client, service["service_id"]
)
await service["coordinator"].async_config_entry_first_refresh()
# Setup the integration
hass.data.setdefault(DOMAIN, {})[entry.entry_id] = {
"client": client,
"services": services,
}
entry.runtime_data = services
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True
async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
async def async_unload_entry(
hass: HomeAssistant, entry: AussieBroadbandConfigEntry
) -> bool:
"""Unload the config entry."""
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
if unload_ok:
hass.data[DOMAIN].pop(entry.entry_id)
return unload_ok
return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)

View File

@ -1,6 +1,8 @@
"""Constants for the Aussie Broadband integration."""
from typing import Final
DEFAULT_UPDATE_INTERVAL = 30
DOMAIN = "aussie_broadband"
SERVICE_ID = "service_id"
SERVICE_ID: Final = "service_id"
CONF_SERVICES = "services"

View File

@ -4,11 +4,12 @@ from __future__ import annotations
from datetime import timedelta
import logging
from typing import Any
from typing import Any, TypedDict
from aussiebb.asyncio import AussieBB
from aussiebb.exceptions import UnrecognisedServiceType
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
@ -17,7 +18,20 @@ from .const import DEFAULT_UPDATE_INTERVAL
_LOGGER = logging.getLogger(__name__)
class AussieBroadandDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
class AussieBroadbandServiceData(TypedDict, total=False):
"""Aussie Broadband service information, extended with the coordinator."""
coordinator: AussieBroadbandDataUpdateCoordinator
description: str
name: str
service_id: str
type: str
type AussieBroadbandConfigEntry = ConfigEntry[list[AussieBroadbandServiceData]]
class AussieBroadbandDataUpdateCoordinator(DataUpdateCoordinator[dict[str, Any]]):
"""Aussie Broadand data update coordinator."""
def __init__(self, hass: HomeAssistant, client: AussieBB, service_id: str) -> None:

View File

@ -5,16 +5,15 @@ from __future__ import annotations
from typing import Any
from homeassistant.components.diagnostics import async_redact_data
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from .const import DOMAIN
from .coordinator import AussieBroadbandConfigEntry
TO_REDACT = ["address", "ipAddresses", "description", "discounts", "coordinator"]
async def async_get_config_entry_diagnostics(
hass: HomeAssistant, config_entry: ConfigEntry
hass: HomeAssistant, config_entry: AussieBroadbandConfigEntry
) -> dict[str, Any]:
"""Return diagnostics for a config entry."""
return {
@ -23,6 +22,6 @@ async def async_get_config_entry_diagnostics(
"service": async_redact_data(service, TO_REDACT),
"usage": async_redact_data(service["coordinator"].data, ["historical"]),
}
for service in hass.data[DOMAIN][config_entry.entry_id]["services"]
for service in config_entry.runtime_data
]
}

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import re
from typing import Any, cast
from typing import cast
from homeassistant.components.sensor import (
SensorDeviceClass,
@ -13,7 +13,6 @@ from homeassistant.components.sensor import (
SensorEntityDescription,
SensorStateClass,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import UnitOfInformation, UnitOfTime
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
@ -22,7 +21,11 @@ from homeassistant.helpers.typing import StateType
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN, SERVICE_ID
from .coordinator import AussieBroadandDataUpdateCoordinator
from .coordinator import (
AussieBroadbandConfigEntry,
AussieBroadbandDataUpdateCoordinator,
AussieBroadbandServiceData,
)
@dataclass(frozen=True)
@ -118,14 +121,16 @@ SENSOR_DESCRIPTIONS: tuple[SensorValueEntityDescription, ...] = (
async def async_setup_entry(
hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
hass: HomeAssistant,
entry: AussieBroadbandConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up the Aussie Broadband sensor platform from a config entry."""
async_add_entities(
[
AussieBroadandSensorEntity(service, description)
for service in hass.data[DOMAIN][entry.entry_id]["services"]
for service in entry.runtime_data
for description in SENSOR_DESCRIPTIONS
if description.key in service["coordinator"].data
]
@ -133,7 +138,7 @@ async def async_setup_entry(
class AussieBroadandSensorEntity(
CoordinatorEntity[AussieBroadandDataUpdateCoordinator], SensorEntity
CoordinatorEntity[AussieBroadbandDataUpdateCoordinator], SensorEntity
):
"""Base class for Aussie Broadband metric sensors."""
@ -141,7 +146,9 @@ class AussieBroadandSensorEntity(
entity_description: SensorValueEntityDescription
def __init__(
self, service: dict[str, Any], description: SensorValueEntityDescription
self,
service: AussieBroadbandServiceData,
description: SensorValueEntityDescription,
) -> None:
"""Initialize the sensor."""
super().__init__(service["coordinator"])