Move juicenet coordinator to separate module (#147168)

This commit is contained in:
epenet 2025-06-20 10:33:17 +02:00 committed by GitHub
parent fde36d5034
commit 84e9422254
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 74 additions and 43 deletions

View File

@ -1,6 +1,5 @@
"""The JuiceNet integration."""
from datetime import timedelta
import logging
import aiohttp
@ -14,9 +13,9 @@ from homeassistant.exceptions import ConfigEntryNotReady
from homeassistant.helpers import config_validation as cv
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from homeassistant.helpers.typing import ConfigType
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, JUICENET_API, JUICENET_COORDINATOR
from .coordinator import JuiceNetCoordinator
from .device import JuiceNetApi
_LOGGER = logging.getLogger(__name__)
@ -74,20 +73,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
return False
_LOGGER.debug("%d JuiceNet device(s) found", len(juicenet.devices))
async def async_update_data():
"""Update all device states from the JuiceNet API."""
for device in juicenet.devices:
await device.update_state(True)
return True
coordinator = DataUpdateCoordinator(
hass,
_LOGGER,
config_entry=entry,
name="JuiceNet",
update_method=async_update_data,
update_interval=timedelta(seconds=30),
)
coordinator = JuiceNetCoordinator(hass, entry, juicenet)
await coordinator.async_config_entry_first_refresh()

View File

@ -0,0 +1,33 @@
"""The JuiceNet integration."""
from datetime import timedelta
import logging
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .device import JuiceNetApi
_LOGGER = logging.getLogger(__name__)
class JuiceNetCoordinator(DataUpdateCoordinator[None]):
"""Coordinator for JuiceNet."""
def __init__(
self, hass: HomeAssistant, entry: ConfigEntry, juicenet_api: JuiceNetApi
) -> None:
"""Initialize the JuiceNet coordinator."""
super().__init__(
hass,
_LOGGER,
config_entry=entry,
name="JuiceNet",
update_interval=timedelta(seconds=30),
)
self.juicenet_api = juicenet_api
async def _async_update_data(self) -> None:
for device in self.juicenet_api.devices:
await device.update_state(True)

View File

@ -1,19 +1,21 @@
"""Adapter to wrap the pyjuicenet api for home assistant."""
from pyjuicenet import Api, Charger
class JuiceNetApi:
"""Represent a connection to JuiceNet."""
def __init__(self, api):
def __init__(self, api: Api) -> None:
"""Create an object from the provided API instance."""
self.api = api
self._devices = []
self._devices: list[Charger] = []
async def setup(self):
async def setup(self) -> None:
"""JuiceNet device setup."""
self._devices = await self.api.get_devices()
@property
def devices(self) -> list:
def devices(self) -> list[Charger]:
"""Get a list of devices managed by this account."""
return self._devices

View File

@ -3,21 +3,19 @@
from pyjuicenet import Charger
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import (
CoordinatorEntity,
DataUpdateCoordinator,
)
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import JuiceNetCoordinator
class JuiceNetDevice(CoordinatorEntity):
class JuiceNetEntity(CoordinatorEntity[JuiceNetCoordinator]):
"""Represent a base JuiceNet device."""
_attr_has_entity_name = True
def __init__(
self, device: Charger, key: str, coordinator: DataUpdateCoordinator
self, device: Charger, key: str, coordinator: JuiceNetCoordinator
) -> None:
"""Initialise the sensor."""
super().__init__(coordinator)

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass
from pyjuicenet import Api, Charger
from pyjuicenet import Charger
from homeassistant.components.number import (
DEFAULT_MAX_VALUE,
@ -14,10 +14,11 @@ from homeassistant.components.number import (
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator
from .const import DOMAIN, JUICENET_API, JUICENET_COORDINATOR
from .entity import JuiceNetDevice
from .coordinator import JuiceNetCoordinator
from .device import JuiceNetApi
from .entity import JuiceNetEntity
@dataclass(frozen=True, kw_only=True)
@ -47,8 +48,8 @@ async def async_setup_entry(
) -> None:
"""Set up the JuiceNet Numbers."""
juicenet_data = hass.data[DOMAIN][config_entry.entry_id]
api: Api = juicenet_data[JUICENET_API]
coordinator = juicenet_data[JUICENET_COORDINATOR]
api: JuiceNetApi = juicenet_data[JUICENET_API]
coordinator: JuiceNetCoordinator = juicenet_data[JUICENET_COORDINATOR]
entities = [
JuiceNetNumber(device, description, coordinator)
@ -58,7 +59,7 @@ async def async_setup_entry(
async_add_entities(entities)
class JuiceNetNumber(JuiceNetDevice, NumberEntity):
class JuiceNetNumber(JuiceNetEntity, NumberEntity):
"""Implementation of a JuiceNet number."""
entity_description: JuiceNetNumberEntityDescription
@ -67,7 +68,7 @@ class JuiceNetNumber(JuiceNetDevice, NumberEntity):
self,
device: Charger,
description: JuiceNetNumberEntityDescription,
coordinator: DataUpdateCoordinator,
coordinator: JuiceNetCoordinator,
) -> None:
"""Initialise the number."""
super().__init__(device, description.key, coordinator)

View File

@ -2,6 +2,8 @@
from __future__ import annotations
from pyjuicenet import Charger
from homeassistant.components.sensor import (
SensorDeviceClass,
SensorEntity,
@ -21,7 +23,9 @@ from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, JUICENET_API, JUICENET_COORDINATOR
from .entity import JuiceNetDevice
from .coordinator import JuiceNetCoordinator
from .device import JuiceNetApi
from .entity import JuiceNetEntity
SENSOR_TYPES: tuple[SensorEntityDescription, ...] = (
SensorEntityDescription(
@ -74,8 +78,8 @@ async def async_setup_entry(
) -> None:
"""Set up the JuiceNet Sensors."""
juicenet_data = hass.data[DOMAIN][config_entry.entry_id]
api = juicenet_data[JUICENET_API]
coordinator = juicenet_data[JUICENET_COORDINATOR]
api: JuiceNetApi = juicenet_data[JUICENET_API]
coordinator: JuiceNetCoordinator = juicenet_data[JUICENET_COORDINATOR]
entities = [
JuiceNetSensorDevice(device, coordinator, description)
@ -85,11 +89,14 @@ async def async_setup_entry(
async_add_entities(entities)
class JuiceNetSensorDevice(JuiceNetDevice, SensorEntity):
class JuiceNetSensorDevice(JuiceNetEntity, SensorEntity):
"""Implementation of a JuiceNet sensor."""
def __init__(
self, device, coordinator, description: SensorEntityDescription
self,
device: Charger,
coordinator: JuiceNetCoordinator,
description: SensorEntityDescription,
) -> None:
"""Initialise the sensor."""
super().__init__(device, description.key, coordinator)

View File

@ -2,13 +2,17 @@
from typing import Any
from pyjuicenet import Charger
from homeassistant.components.switch import SwitchEntity
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .const import DOMAIN, JUICENET_API, JUICENET_COORDINATOR
from .entity import JuiceNetDevice
from .coordinator import JuiceNetCoordinator
from .device import JuiceNetApi
from .entity import JuiceNetEntity
async def async_setup_entry(
@ -18,20 +22,20 @@ async def async_setup_entry(
) -> None:
"""Set up the JuiceNet switches."""
juicenet_data = hass.data[DOMAIN][config_entry.entry_id]
api = juicenet_data[JUICENET_API]
coordinator = juicenet_data[JUICENET_COORDINATOR]
api: JuiceNetApi = juicenet_data[JUICENET_API]
coordinator: JuiceNetCoordinator = juicenet_data[JUICENET_COORDINATOR]
async_add_entities(
JuiceNetChargeNowSwitch(device, coordinator) for device in api.devices
)
class JuiceNetChargeNowSwitch(JuiceNetDevice, SwitchEntity):
class JuiceNetChargeNowSwitch(JuiceNetEntity, SwitchEntity):
"""Implementation of a JuiceNet switch."""
_attr_translation_key = "charge_now"
def __init__(self, device, coordinator):
def __init__(self, device: Charger, coordinator: JuiceNetCoordinator) -> None:
"""Initialise the switch."""
super().__init__(device, "charge_now", coordinator)