Add update platform to Peblar Rocksolid EV Chargers integration (#133570)

* Add update platform to Peblar Rocksolid EV Chargers integration

* Use device class translations
This commit is contained in:
Franck Nijhof 2024-12-21 10:55:00 +01:00 committed by GitHub
parent 4a063c3f9e
commit 859993e443
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 229 additions and 39 deletions

View File

@ -2,6 +2,8 @@
from __future__ import annotations
import asyncio
from aiohttp import CookieJar
from peblar import (
AccessMode,
@ -14,22 +16,34 @@ from peblar import (
from homeassistant.const import CONF_HOST, CONF_PASSWORD, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import ConfigEntryError, ConfigEntryNotReady
from homeassistant.helpers import device_registry as dr
from homeassistant.helpers.aiohttp_client import async_create_clientsession
from .coordinator import PeblarConfigEntry, PeblarMeterDataUpdateCoordinator
from .const import DOMAIN
from .coordinator import (
PeblarConfigEntry,
PeblarMeterDataUpdateCoordinator,
PeblarRuntimeData,
PeblarVersionDataUpdateCoordinator,
)
PLATFORMS = [Platform.SENSOR]
PLATFORMS = [
Platform.SENSOR,
Platform.UPDATE,
]
async def async_setup_entry(hass: HomeAssistant, entry: PeblarConfigEntry) -> bool:
"""Set up Peblar from a config entry."""
# Set up connection to the Peblar charger
peblar = Peblar(
host=entry.data[CONF_HOST],
session=async_create_clientsession(hass, cookie_jar=CookieJar(unsafe=True)),
)
try:
await peblar.login(password=entry.data[CONF_PASSWORD])
system_information = await peblar.system_information()
api = await peblar.rest_api(enable=True, access_mode=AccessMode.READ_WRITE)
except PeblarConnectionError as err:
raise ConfigEntryNotReady("Could not connect to Peblar charger") from err
@ -40,10 +54,41 @@ async def async_setup_entry(hass: HomeAssistant, entry: PeblarConfigEntry) -> bo
"Unknown error occurred while connecting to Peblar charger"
) from err
coordinator = PeblarMeterDataUpdateCoordinator(hass, entry, api)
await coordinator.async_config_entry_first_refresh()
# Setup the data coordinators
meter_coordinator = PeblarMeterDataUpdateCoordinator(hass, entry, api)
version_coordinator = PeblarVersionDataUpdateCoordinator(hass, entry, peblar)
await asyncio.gather(
meter_coordinator.async_config_entry_first_refresh(),
version_coordinator.async_config_entry_first_refresh(),
)
entry.runtime_data = coordinator
# Store the runtime data
entry.runtime_data = PeblarRuntimeData(
system_information=system_information,
meter_coordinator=meter_coordinator,
version_coordinator=version_coordinator,
)
# Peblar is a single device integration. Setting up the device directly
# during setup. This way we only have to reference it in all entities.
device_registry = dr.async_get(hass)
device_registry.async_get_or_create(
config_entry_id=entry.entry_id,
configuration_url=f"http://{entry.data[CONF_HOST]}",
connections={
(dr.CONNECTION_NETWORK_MAC, system_information.ethernet_mac_address),
(dr.CONNECTION_NETWORK_MAC, system_information.wlan_mac_address),
},
identifiers={(DOMAIN, system_information.product_serial_number)},
manufacturer=system_information.product_vendor_name,
model_id=system_information.product_number,
model=system_information.product_model_name,
name="Peblar EV Charger",
serial_number=system_information.product_serial_number,
sw_version=version_coordinator.data.current.firmware,
)
# Forward the setup to the platforms
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
return True

View File

@ -1,16 +1,67 @@
"""Data update coordinator for Peblar EV chargers."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import timedelta
from peblar import PeblarApi, PeblarError, PeblarMeter
from peblar import Peblar, PeblarApi, PeblarError, PeblarMeter, PeblarVersions
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
from tests.components.peblar.conftest import PeblarSystemInformation
from .const import LOGGER
type PeblarConfigEntry = ConfigEntry[PeblarMeterDataUpdateCoordinator]
@dataclass(kw_only=True)
class PeblarRuntimeData:
"""Class to hold runtime data."""
system_information: PeblarSystemInformation
meter_coordinator: PeblarMeterDataUpdateCoordinator
version_coordinator: PeblarVersionDataUpdateCoordinator
type PeblarConfigEntry = ConfigEntry[PeblarRuntimeData]
@dataclass(kw_only=True, frozen=True)
class PeblarVersionInformation:
"""Class to hold version information."""
current: PeblarVersions
available: PeblarVersions
class PeblarVersionDataUpdateCoordinator(
DataUpdateCoordinator[PeblarVersionInformation]
):
"""Class to manage fetching Peblar version information."""
def __init__(
self, hass: HomeAssistant, entry: PeblarConfigEntry, peblar: Peblar
) -> None:
"""Initialize the coordinator."""
self.peblar = peblar
super().__init__(
hass,
LOGGER,
config_entry=entry,
name=f"Peblar {entry.title} version",
update_interval=timedelta(hours=2),
)
async def _async_update_data(self) -> PeblarVersionInformation:
"""Fetch data from the Peblar device."""
try:
return PeblarVersionInformation(
current=await self.peblar.current_versions(),
available=await self.peblar.available_versions(),
)
except PeblarError as err:
raise UpdateFailed(err) from err
class PeblarMeterDataUpdateCoordinator(DataUpdateCoordinator[PeblarMeter]):

View File

@ -1,26 +0,0 @@
"""Base entity for the Peblar integration."""
from __future__ import annotations
from homeassistant.const import CONF_HOST
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import PeblarConfigEntry, PeblarMeterDataUpdateCoordinator
class PeblarEntity(CoordinatorEntity[PeblarMeterDataUpdateCoordinator]):
"""Defines a Peblar entity."""
_attr_has_entity_name = True
def __init__(self, entry: PeblarConfigEntry) -> None:
"""Initialize the Peblar entity."""
super().__init__(coordinator=entry.runtime_data)
self._attr_device_info = DeviceInfo(
configuration_url=f"http://{entry.data[CONF_HOST]}",
identifiers={(DOMAIN, str(entry.unique_id))},
manufacturer="Peblar",
name="Peblar EV charger",
)

View File

@ -0,0 +1,9 @@
{
"entity": {
"update": {
"customization": {
"default": "mdi:palette"
}
}
}
}

View File

@ -15,10 +15,12 @@ from homeassistant.components.sensor import (
)
from homeassistant.const import UnitOfEnergy
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .coordinator import PeblarConfigEntry
from .entity import PeblarEntity
from .const import DOMAIN
from .coordinator import PeblarConfigEntry, PeblarMeterDataUpdateCoordinator
@dataclass(frozen=True, kw_only=True)
@ -28,7 +30,7 @@ class PeblarSensorDescription(SensorEntityDescription):
value_fn: Callable[[PeblarMeter], int | None]
SENSORS: tuple[PeblarSensorDescription, ...] = (
DESCRIPTIONS: tuple[PeblarSensorDescription, ...] = (
PeblarSensorDescription(
key="energy_total",
device_class=SensorDeviceClass.ENERGY,
@ -48,24 +50,33 @@ async def async_setup_entry(
) -> None:
"""Set up Peblar sensors based on a config entry."""
async_add_entities(
PeblarSensorEntity(entry, description) for description in SENSORS
PeblarSensorEntity(entry, description) for description in DESCRIPTIONS
)
class PeblarSensorEntity(PeblarEntity, SensorEntity):
class PeblarSensorEntity(
CoordinatorEntity[PeblarMeterDataUpdateCoordinator], SensorEntity
):
"""Defines a Peblar sensor."""
entity_description: PeblarSensorDescription
_attr_has_entity_name = True
def __init__(
self,
entry: PeblarConfigEntry,
description: PeblarSensorDescription,
) -> None:
"""Initialize the Peblar entity."""
super().__init__(entry)
super().__init__(entry.runtime_data.meter_coordinator)
self.entity_description = description
self._attr_unique_id = f"{entry.unique_id}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={
(DOMAIN, entry.runtime_data.system_information.product_serial_number)
},
)
@property
def native_value(self) -> int | None:

View File

@ -31,5 +31,12 @@
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]",
"no_serial_number": "The discovered Peblar device did not provide a serial number."
}
},
"entity": {
"update": {
"customization": {
"name": "Customization"
}
}
}
}

View File

@ -0,0 +1,93 @@
"""Support for Peblar updates."""
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
from homeassistant.components.update import (
UpdateDeviceClass,
UpdateEntity,
UpdateEntityDescription,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.update_coordinator import CoordinatorEntity
from .const import DOMAIN
from .coordinator import (
PeblarConfigEntry,
PeblarVersionDataUpdateCoordinator,
PeblarVersionInformation,
)
@dataclass(frozen=True, kw_only=True)
class PeblarUpdateEntityDescription(UpdateEntityDescription):
"""Describe an Peblar update entity."""
installed_fn: Callable[[PeblarVersionInformation], str | None]
available_fn: Callable[[PeblarVersionInformation], str | None]
DESCRIPTIONS: tuple[PeblarUpdateEntityDescription, ...] = (
PeblarUpdateEntityDescription(
key="firmware",
device_class=UpdateDeviceClass.FIRMWARE,
installed_fn=lambda x: x.current.firmware,
available_fn=lambda x: x.available.firmware,
),
PeblarUpdateEntityDescription(
key="customization",
translation_key="customization",
installed_fn=lambda x: x.current.customization,
available_fn=lambda x: x.available.customization,
),
)
async def async_setup_entry(
hass: HomeAssistant,
entry: PeblarConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Peblar update based on a config entry."""
async_add_entities(
PeblarUpdateEntity(entry, description) for description in DESCRIPTIONS
)
class PeblarUpdateEntity(
CoordinatorEntity[PeblarVersionDataUpdateCoordinator], UpdateEntity
):
"""Defines a Peblar update entity."""
entity_description: PeblarUpdateEntityDescription
_attr_has_entity_name = True
def __init__(
self,
entry: PeblarConfigEntry,
description: PeblarUpdateEntityDescription,
) -> None:
"""Initialize the update entity."""
super().__init__(entry.runtime_data.version_coordinator)
self.entity_description = description
self._attr_unique_id = f"{entry.unique_id}_{description.key}"
self._attr_device_info = DeviceInfo(
identifiers={
(DOMAIN, entry.runtime_data.system_information.product_serial_number)
},
)
@property
def installed_version(self) -> str | None:
"""Version currently installed and in use."""
return self.entity_description.installed_fn(self.coordinator.data)
@property
def latest_version(self) -> str | None:
"""Latest version available for install."""
return self.entity_description.available_fn(self.coordinator.data)