Implement battery group mode in HomeWizard (#146770)

* Implement battery group mode for HomeWizard P1

* Clean up test

* Disable 'entity_registry_enabled_default'

* Fix failing tests because of 'entity_registry_enabled_default'

* Proof entities are disabled by default

* Undo dev change

* Update homeassistant/components/homewizard/select.py

* Update homeassistant/components/homewizard/select.py

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>

* Update homeassistant/components/homewizard/strings.json

* Apply suggestions from code review

* Update tests due to updated translations

---------

Co-authored-by: Joost Lekkerkerker <joostlek@outlook.com>
This commit is contained in:
Duco Sebel 2025-06-20 13:19:45 +02:00 committed by GitHub
parent 9ae9ad1e43
commit a493bdc208
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 540 additions and 8 deletions

View File

@ -8,7 +8,13 @@ import logging
from homeassistant.const import Platform
DOMAIN = "homewizard"
PLATFORMS = [Platform.BUTTON, Platform.NUMBER, Platform.SENSOR, Platform.SWITCH]
PLATFORMS = [
Platform.BUTTON,
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
]
LOGGER = logging.getLogger(__package__)

View File

@ -5,7 +5,7 @@ from __future__ import annotations
from collections.abc import Callable, Coroutine
from typing import Any, Concatenate
from homewizard_energy.errors import DisabledError, RequestError
from homewizard_energy.errors import DisabledError, RequestError, UnauthorizedError
from homeassistant.exceptions import HomeAssistantError
@ -41,5 +41,10 @@ def homewizard_exception_handler[_HomeWizardEntityT: HomeWizardEntity, **_P](
translation_domain=DOMAIN,
translation_key="api_disabled",
) from ex
except UnauthorizedError as ex:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="api_unauthorized",
) from ex
return handler

View File

@ -0,0 +1,89 @@
"""Support for HomeWizard select platform."""
from __future__ import annotations
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from typing import Any
from homewizard_energy import HomeWizardEnergy
from homewizard_energy.models import Batteries, CombinedModels as DeviceResponseEntry
from homeassistant.components.select import SelectEntity, SelectEntityDescription
from homeassistant.const import EntityCategory
from homeassistant.core import HomeAssistant
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from .coordinator import HomeWizardConfigEntry, HWEnergyDeviceUpdateCoordinator
from .entity import HomeWizardEntity
from .helpers import homewizard_exception_handler
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class HomeWizardSelectEntityDescription(SelectEntityDescription):
"""Class describing HomeWizard select entities."""
available_fn: Callable[[DeviceResponseEntry], bool]
create_fn: Callable[[DeviceResponseEntry], bool]
current_fn: Callable[[DeviceResponseEntry], str | None]
set_fn: Callable[[HomeWizardEnergy, str], Awaitable[Any]]
DESCRIPTIONS = [
HomeWizardSelectEntityDescription(
key="battery_group_mode",
translation_key="battery_group_mode",
entity_category=EntityCategory.CONFIG,
entity_registry_enabled_default=False,
options=[Batteries.Mode.ZERO, Batteries.Mode.STANDBY, Batteries.Mode.TO_FULL],
available_fn=lambda x: x.batteries is not None,
create_fn=lambda x: x.batteries is not None,
current_fn=lambda x: x.batteries.mode if x.batteries else None,
set_fn=lambda api, mode: api.batteries(mode=Batteries.Mode(mode)),
),
]
async def async_setup_entry(
hass: HomeAssistant,
entry: HomeWizardConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up HomeWizard select based on a config entry."""
async_add_entities(
HomeWizardSelectEntity(
coordinator=entry.runtime_data,
description=description,
)
for description in DESCRIPTIONS
if description.create_fn(entry.runtime_data.data)
)
class HomeWizardSelectEntity(HomeWizardEntity, SelectEntity):
"""Defines a HomeWizard select entity."""
entity_description: HomeWizardSelectEntityDescription
def __init__(
self,
coordinator: HWEnergyDeviceUpdateCoordinator,
description: HomeWizardSelectEntityDescription,
) -> None:
"""Initialize the switch."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = f"{coordinator.config_entry.unique_id}_{description.key}"
@property
def current_option(self) -> str | None:
"""Return the selected entity option to represent the entity state."""
return self.entity_description.current_fn(self.coordinator.data)
@homewizard_exception_handler
async def async_select_option(self, option: str) -> None:
"""Change the selected option."""
await self.entity_description.set_fn(self.coordinator.api, option)
await self.coordinator.async_request_refresh()

View File

@ -152,14 +152,27 @@
"cloud_connection": {
"name": "Cloud connection"
}
},
"select": {
"battery_group_mode": {
"name": "Battery group mode",
"state": {
"zero": "Zero mode",
"to_full": "Manual charge mode",
"standby": "Standby"
}
}
}
},
"exceptions": {
"api_disabled": {
"message": "The local API is disabled."
},
"api_unauthorized": {
"message": "The local API is unauthorized. Restore API access by following the instructions in the repair issue."
},
"communication_error": {
"message": "An error occurred while communicating with HomeWizard device"
"message": "An error occurred while communicating with your HomeWizard Energy device"
}
},
"issues": {

View File

@ -4,6 +4,7 @@ from collections.abc import Generator
from unittest.mock import AsyncMock, MagicMock, patch
from homewizard_energy.models import (
Batteries,
CombinedModels,
Device,
Measurement,
@ -64,6 +65,13 @@ def mock_homewizardenergy(
if get_fixture_path(f"{device_fixture}/system.json", DOMAIN).exists()
else None
),
batteries=(
Batteries.from_dict(
load_json_object_fixture(f"{device_fixture}/batteries.json", DOMAIN)
)
if get_fixture_path(f"{device_fixture}/batteries.json", DOMAIN).exists()
else None
),
)
# device() call is used during configuration flow
@ -112,6 +120,13 @@ def mock_homewizardenergy_v2(
if get_fixture_path(f"v2/{device_fixture}/system.json", DOMAIN).exists()
else None
),
batteries=(
Batteries.from_dict(
load_json_object_fixture(f"{device_fixture}/batteries.json", DOMAIN)
)
if get_fixture_path(f"{device_fixture}/batteries.json", DOMAIN).exists()
else None
),
)
# device() call is used during configuration flow

View File

@ -0,0 +1,7 @@
{
"mode": "zero",
"power_w": -404,
"target_power_w": -400,
"max_consumption_w": 1600,
"max_production_w": 800
}

View File

@ -278,7 +278,13 @@
# name: test_diagnostics[HWE-P1]
dict({
'data': dict({
'batteries': None,
'batteries': dict({
'max_consumption_w': 1600.0,
'max_production_w': 800.0,
'mode': 'zero',
'power_w': -404.0,
'target_power_w': -400.0,
}),
'device': dict({
'api_version': '1.0.0',
'firmware_version': '4.19',

View File

@ -0,0 +1,97 @@
# serializer version: 1
# name: test_select_entity_snapshots[HWE-P1-select.device_battery_group_mode]
StateSnapshot({
'attributes': ReadOnlyDict({
'friendly_name': 'Device Battery group mode',
'options': list([
<Mode.ZERO: 'zero'>,
<Mode.STANDBY: 'standby'>,
<Mode.TO_FULL: 'to_full'>,
]),
}),
'context': <ANY>,
'entity_id': 'select.device_battery_group_mode',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'zero',
})
# ---
# name: test_select_entity_snapshots[HWE-P1-select.device_battery_group_mode].1
EntityRegistryEntrySnapshot({
'aliases': set({
}),
'area_id': None,
'capabilities': dict({
'options': list([
<Mode.ZERO: 'zero'>,
<Mode.STANDBY: 'standby'>,
<Mode.TO_FULL: 'to_full'>,
]),
}),
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'select',
'entity_category': <EntityCategory.CONFIG: 'config'>,
'entity_id': 'select.device_battery_group_mode',
'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': 'Battery group mode',
'platform': 'homewizard',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'battery_group_mode',
'unique_id': 'HWE-P1_5c2fafabcdef_battery_group_mode',
'unit_of_measurement': None,
})
# ---
# name: test_select_entity_snapshots[HWE-P1-select.device_battery_group_mode].2
DeviceRegistryEntrySnapshot({
'area_id': None,
'config_entries': <ANY>,
'config_entries_subentries': <ANY>,
'configuration_url': None,
'connections': set({
tuple(
'mac',
'5c:2f:af:ab:cd:ef',
),
}),
'disabled_by': None,
'entry_type': None,
'hw_version': None,
'id': <ANY>,
'identifiers': set({
tuple(
'homewizard',
'5c2fafabcdef',
),
}),
'is_new': False,
'labels': set({
}),
'manufacturer': 'HomeWizard',
'model': 'Wi-Fi P1 Meter',
'model_id': 'HWE-P1',
'name': 'Device',
'name_by_user': None,
'primary_config_entry': <ANY>,
'serial_number': None,
'suggested_area': None,
'sw_version': '4.19',
'via_device_id': None,
})
# ---

View File

@ -61,7 +61,7 @@ async def test_identify_button(
with pytest.raises(
HomeAssistantError,
match=r"^An error occurred while communicating with HomeWizard device$",
match=r"^An error occurred while communicating with your HomeWizard Energy device$",
):
await hass.services.async_call(
button.DOMAIN,

View File

@ -73,7 +73,7 @@ async def test_number_entities(
mock_homewizardenergy.system.side_effect = RequestError
with pytest.raises(
HomeAssistantError,
match=r"^An error occurred while communicating with HomeWizard device$",
match=r"^An error occurred while communicating with your HomeWizard Energy device$",
):
await hass.services.async_call(
number.DOMAIN,

View File

@ -0,0 +1,294 @@
"""Test the Select entity for HomeWizard."""
from unittest.mock import MagicMock
from homewizard_energy import UnsupportedError
from homewizard_energy.errors import RequestError, UnauthorizedError
from homewizard_energy.models import Batteries
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.homewizard.const import UPDATE_INTERVAL
from homeassistant.components.select import (
ATTR_OPTION,
DOMAIN as SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_UNAVAILABLE
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import device_registry as dr, entity_registry as er
from homeassistant.util import dt as dt_util
from tests.common import async_fire_time_changed
pytestmark = [
pytest.mark.usefixtures("init_integration"),
]
@pytest.mark.parametrize(
("device_fixture", "entity_ids"),
[
(
"HWE-WTR",
[
"select.device_battery_group_mode",
],
),
(
"SDM230",
[
"select.device_battery_group_mode",
],
),
(
"SDM630",
[
"select.device_battery_group_mode",
],
),
(
"HWE-KWH1",
[
"select.device_battery_group_mode",
],
),
(
"HWE-KWH3",
[
"select.device_battery_group_mode",
],
),
],
)
async def test_entities_not_created_for_device(
hass: HomeAssistant,
entity_ids: list[str],
) -> None:
"""Ensures entities for a specific device are not created."""
for entity_id in entity_ids:
assert not hass.states.get(entity_id)
@pytest.mark.parametrize(
("device_fixture", "entity_id"),
[
("HWE-P1", "select.device_battery_group_mode"),
],
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_select_entity_snapshots(
hass: HomeAssistant,
device_registry: dr.DeviceRegistry,
entity_registry: er.EntityRegistry,
snapshot: SnapshotAssertion,
entity_id: str,
) -> None:
"""Test that select entity state and registry entries match snapshots."""
assert (state := hass.states.get(entity_id))
assert snapshot == state
assert (entity_entry := entity_registry.async_get(entity_id))
assert snapshot == entity_entry
assert entity_entry.device_id
assert (device_entry := device_registry.async_get(entity_entry.device_id))
assert snapshot == device_entry
@pytest.mark.parametrize(
("device_fixture", "entity_id", "option", "expected_mode"),
[
(
"HWE-P1",
"select.device_battery_group_mode",
"standby",
Batteries.Mode.STANDBY,
),
(
"HWE-P1",
"select.device_battery_group_mode",
"to_full",
Batteries.Mode.TO_FULL,
),
("HWE-P1", "select.device_battery_group_mode", "zero", Batteries.Mode.ZERO),
],
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_select_set_option(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
entity_id: str,
option: str,
expected_mode: Batteries.Mode,
) -> None:
"""Test that selecting an option calls the correct API method."""
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: option,
},
blocking=True,
)
mock_homewizardenergy.batteries.assert_called_with(mode=expected_mode)
@pytest.mark.parametrize(
("device_fixture", "entity_id", "option"),
[
("HWE-P1", "select.device_battery_group_mode", "zero"),
("HWE-P1", "select.device_battery_group_mode", "standby"),
("HWE-P1", "select.device_battery_group_mode", "to_full"),
],
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_select_request_error(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
entity_id: str,
option: str,
) -> None:
"""Test that RequestError is handled and raises HomeAssistantError."""
mock_homewizardenergy.batteries.side_effect = RequestError
with pytest.raises(
HomeAssistantError,
match=r"^An error occurred while communicating with your HomeWizard Energy device$",
):
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: option,
},
blocking=True,
)
@pytest.mark.parametrize(
("device_fixture", "entity_id", "option"),
[
("HWE-P1", "select.device_battery_group_mode", "to_full"),
],
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_select_unauthorized_error(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
entity_id: str,
option: str,
) -> None:
"""Test that UnauthorizedError is handled and raises HomeAssistantError."""
mock_homewizardenergy.batteries.side_effect = UnauthorizedError
with pytest.raises(
HomeAssistantError,
match=r"^The local API is unauthorized\. Restore API access by following the instructions in the repair issue$",
):
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: option,
},
blocking=True,
)
@pytest.mark.parametrize("device_fixture", ["HWE-P1"])
@pytest.mark.parametrize("exception", [RequestError, UnsupportedError])
@pytest.mark.parametrize(
("entity_id", "method"),
[
("select.device_battery_group_mode", "combined"),
],
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_select_unreachable(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
exception: Exception,
entity_id: str,
method: str,
) -> None:
"""Test that unreachable devices are marked as unavailable."""
mocked_method = getattr(mock_homewizardenergy, method)
mocked_method.side_effect = exception
async_fire_time_changed(hass, dt_util.utcnow() + UPDATE_INTERVAL)
await hass.async_block_till_done()
assert (state := hass.states.get(entity_id))
assert state.state == STATE_UNAVAILABLE
@pytest.mark.parametrize(
("device_fixture", "entity_id"),
[
("HWE-P1", "select.device_battery_group_mode"),
],
)
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
async def test_select_multiple_state_changes(
hass: HomeAssistant,
mock_homewizardenergy: MagicMock,
entity_id: str,
) -> None:
"""Test changing select state multiple times in sequence."""
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "zero",
},
blocking=True,
)
mock_homewizardenergy.batteries.assert_called_with(mode=Batteries.Mode.ZERO)
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "to_full",
},
blocking=True,
)
mock_homewizardenergy.batteries.assert_called_with(mode=Batteries.Mode.TO_FULL)
await hass.services.async_call(
SELECT_DOMAIN,
SERVICE_SELECT_OPTION,
{
ATTR_ENTITY_ID: entity_id,
ATTR_OPTION: "standby",
},
blocking=True,
)
mock_homewizardenergy.batteries.assert_called_with(mode=Batteries.Mode.STANDBY)
@pytest.mark.parametrize(
("device_fixture", "entity_ids"),
[
(
"HWE-P1",
[
"select.device_battery_group_mode",
],
),
],
)
async def test_disabled_by_default_selects(
hass: HomeAssistant, entity_registry: er.EntityRegistry, entity_ids: list[str]
) -> None:
"""Test the disabled by default selects."""
for entity_id in entity_ids:
assert not hass.states.get(entity_id)
assert (entry := entity_registry.async_get(entity_id))
assert entry.disabled
assert entry.disabled_by is er.RegistryEntryDisabler.INTEGRATION

View File

@ -149,7 +149,7 @@ async def test_switch_entities(
with pytest.raises(
HomeAssistantError,
match=r"^An error occurred while communicating with HomeWizard device$",
match=r"^An error occurred while communicating with your HomeWizard Energy device$",
):
await hass.services.async_call(
switch.DOMAIN,
@ -160,7 +160,7 @@ async def test_switch_entities(
with pytest.raises(
HomeAssistantError,
match=r"^An error occurred while communicating with HomeWizard device$",
match=r"^An error occurred while communicating with your HomeWizard Energy device$",
):
await hass.services.async_call(
switch.DOMAIN,