Add switch platform to Peblar Rocksolid EV Chargers integration (#133749)

This commit is contained in:
Franck Nijhof 2024-12-21 20:28:55 +01:00 committed by GitHub
parent 04276d3523
commit c67e2047e3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 227 additions and 0 deletions

View File

@ -32,6 +32,7 @@ PLATFORMS = [
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
Platform.UPDATE,
]

View File

@ -11,6 +11,7 @@ from peblar import (
PeblarError,
PeblarEVInterface,
PeblarMeter,
PeblarSystem,
PeblarUserConfiguration,
PeblarVersions,
)
@ -55,6 +56,7 @@ class PeblarData:
ev: PeblarEVInterface
meter: PeblarMeter
system: PeblarSystem
class PeblarVersionDataUpdateCoordinator(
@ -108,6 +110,7 @@ class PeblarDataUpdateCoordinator(DataUpdateCoordinator[PeblarData]):
return PeblarData(
ev=await self.api.ev_interface(),
meter=await self.api.meter(),
system=await self.api.system(),
)
except PeblarError as err:
raise UpdateFailed(err) from err

View File

@ -18,6 +18,7 @@ async def async_get_config_entry_diagnostics(
"user_configuration": entry.runtime_data.user_configuraton_coordinator.data.to_dict(),
"ev": entry.runtime_data.data_coordinator.data.ev.to_dict(),
"meter": entry.runtime_data.data_coordinator.data.meter.to_dict(),
"system": entry.runtime_data.data_coordinator.data.system.to_dict(),
"versions": {
"available": entry.runtime_data.version_coordinator.data.available.to_dict(),
"current": entry.runtime_data.version_coordinator.data.current.to_dict(),

View File

@ -16,6 +16,11 @@
}
}
},
"switch": {
"force_single_phase": {
"default": "mdi:power-cycle"
}
},
"update": {
"customization": {
"default": "mdi:palette"

View File

@ -85,6 +85,11 @@
"name": "Voltage phase 3"
}
},
"switch": {
"force_single_phase": {
"name": "Force single phase"
}
},
"update": {
"customization": {
"name": "Customization"

View File

@ -0,0 +1,102 @@
"""Support for Peblar selects."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any
from peblar import PeblarApi
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.const import EntityCategory
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,
PeblarData,
PeblarDataUpdateCoordinator,
PeblarRuntimeData,
)
@dataclass(frozen=True, kw_only=True)
class PeblarSwitchEntityDescription(SwitchEntityDescription):
"""Class describing Peblar switch entities."""
has_fn: Callable[[PeblarRuntimeData], bool] = lambda x: True
is_on_fn: Callable[[PeblarData], bool]
set_fn: Callable[[PeblarApi, bool], Awaitable[Any]]
DESCRIPTIONS = [
PeblarSwitchEntityDescription(
key="force_single_phase",
translation_key="force_single_phase",
entity_category=EntityCategory.CONFIG,
has_fn=lambda x: (
x.data_coordinator.data.system.force_single_phase_allowed
and x.user_configuraton_coordinator.data.connected_phases > 1
),
is_on_fn=lambda x: x.ev.force_single_phase,
set_fn=lambda x, on: x.ev_interface(force_single_phase=on),
),
]
async def async_setup_entry(
hass: HomeAssistant,
entry: PeblarConfigEntry,
async_add_entities: AddEntitiesCallback,
) -> None:
"""Set up Peblar switch based on a config entry."""
async_add_entities(
PeblarSwitchEntity(
entry=entry,
description=description,
)
for description in DESCRIPTIONS
if description.has_fn(entry.runtime_data)
)
class PeblarSwitchEntity(CoordinatorEntity[PeblarDataUpdateCoordinator], SwitchEntity):
"""Defines a Peblar switch entity."""
entity_description: PeblarSwitchEntityDescription
_attr_has_entity_name = True
def __init__(
self,
entry: PeblarConfigEntry,
description: PeblarSwitchEntityDescription,
) -> None:
"""Initialize the select entity."""
super().__init__(entry.runtime_data.data_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 is_on(self) -> bool:
"""Return state of the switch."""
return self.entity_description.is_on_fn(self.coordinator.data)
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the entity on."""
await self.entity_description.set_fn(self.coordinator.api, True)
await self.coordinator.async_request_refresh()
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the entity off."""
await self.entity_description.set_fn(self.coordinator.api, False)
await self.coordinator.async_request_refresh()

View File

@ -9,6 +9,7 @@ from unittest.mock import MagicMock, patch
from peblar import (
PeblarEVInterface,
PeblarMeter,
PeblarSystem,
PeblarSystemInformation,
PeblarUserConfiguration,
PeblarVersions,
@ -71,6 +72,9 @@ def mock_peblar() -> Generator[MagicMock]:
api.meter.return_value = PeblarMeter.from_json(
load_fixture("meter.json", DOMAIN)
)
api.system.return_value = PeblarSystem.from_json(
load_fixture("system.json", DOMAIN)
)
yield peblar

View File

@ -0,0 +1,12 @@
{
"ActiveErrorCodes": [],
"ActiveWarningCodes": [],
"CellularSignalStrength": null,
"FirmwareVersion": "1.6.1+1+WL-1",
"Force1PhaseAllowed": true,
"PhaseCount": 3,
"ProductPn": "6004-2300-8002",
"ProductSn": "23-45-A4O-MOF",
"Uptime": 322094,
"WlanSignalStrength": null
}

View File

@ -20,6 +20,18 @@
'PowerTotal': 3185,
'VoltagePhase1': 223,
}),
'system': dict({
'ActiveErrorCodes': list([
]),
'ActiveWarningCodes': list([
]),
'FirmwareVersion': '1.6.1+1+WL-1',
'Force1PhaseAllowed': True,
'PhaseCount': 3,
'ProductPn': '6004-2300-8002',
'ProductSn': '23-45-A4O-MOF',
'Uptime': 322094,
}),
'system_information': dict({
'BopCalIGainA': 264625,
'BopCalIGainB': 267139,

View File

@ -0,0 +1,47 @@
# serializer version: 1
# name: test_entities[switch][switch.peblar_ev_charger_force_single_phase-entry]
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'switch.peblar_ev_charger_force_single_phase',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Force single phase',
'platform': 'peblar',
'previous_unique_id': None,
'supported_features': 0,
'translation_key': 'force_single_phase',
'unique_id': '23-45-A4O-MOF-force_single_phase',
'unit_of_measurement': None,
})
# ---
# name: test_entities[switch][switch.peblar_ev_charger_force_single_phase-state]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Peblar EV Charger Force single phase',
}),
'context': <ANY>,
'entity_id': 'switch.peblar_ev_charger_force_single_phase',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---

View File

@ -0,0 +1,35 @@
"""Tests for the Peblar switch platform."""
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.peblar.const import DOMAIN
from homeassistant.const import Platform
from homeassistant.core import HomeAssistant
from homeassistant.helpers import device_registry as dr, entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
@pytest.mark.parametrize("init_integration", [Platform.SWITCH], indirect=True)
@pytest.mark.usefixtures("init_integration")
async def test_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
device_registry: dr.DeviceRegistry,
mock_config_entry: MockConfigEntry,
) -> None:
"""Test the switch entities."""
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
# Ensure all entities are correctly assigned to the Peblar device
device_entry = device_registry.async_get_device(
identifiers={(DOMAIN, "23-45-A4O-MOF")}
)
assert device_entry
entity_entries = er.async_entries_for_config_entry(
entity_registry, mock_config_entry.entry_id
)
for entity_entry in entity_entries:
assert entity_entry.device_id == device_entry.id