Add switch platform to Vistapool (#172540)

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
fdebrus
2026-08-26 18:12:12 +02:00
committed by GitHub
co-authored by Claude
parent 81a59063ae
commit 5587b9883d
5 changed files with 706 additions and 0 deletions
@@ -36,6 +36,7 @@ PLATFORMS: list[Platform] = [
Platform.NUMBER,
Platform.SELECT,
Platform.SENSOR,
Platform.SWITCH,
Platform.TIME,
]
@@ -208,6 +208,26 @@
"name": "UV"
}
},
"switch": {
"electrolysis_boost": {
"name": "Electrolysis boost"
},
"electrolysis_cover": {
"name": "Electrolysis cover"
},
"filtration": {
"name": "Filtration"
},
"heating_climate": {
"name": "Heating climate"
},
"relay": {
"name": "Relay {number}"
},
"smart_mode_freeze": {
"name": "Smart mode freeze"
}
},
"time": {
"filtration_interval_end": {
"name": "Filtration interval {number} end"
@@ -0,0 +1,167 @@
"""Vistapool Switch entities."""
from dataclasses import dataclass
from typing import Any, override
from aioaquarite import AquariteError
from homeassistant.components.switch import SwitchEntity, SwitchEntityDescription
from homeassistant.core import HomeAssistant, callback
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers.dispatcher import async_dispatcher_connect
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
from . import VistapoolConfigEntry
from .const import DOMAIN, PATH_HASHIDRO, SIGNAL_NEW_POOL
from .coordinator import VistapoolDataUpdateCoordinator
from .entity import VistapoolEntity
PARALLEL_UPDATES = 1
@dataclass(frozen=True, kw_only=True)
class VistapoolSwitchEntityDescription(SwitchEntityDescription):
"""Describes a Vistapool switch entity."""
value_path: str
extra_read_paths: tuple[str, ...] = ()
exists_path: str | tuple[str, ...] | None = None
SWITCH_DESCRIPTIONS: tuple[VistapoolSwitchEntityDescription, ...] = (
VistapoolSwitchEntityDescription(
key="filtration",
translation_key="filtration",
value_path="filtration.status",
),
*(
VistapoolSwitchEntityDescription(
key=f"relay_{i}",
translation_key="relay",
translation_placeholders={"number": str(i)},
value_path=f"relays.relay{i}.info.onoff",
extra_read_paths=(f"relays.relay{i}.info.status",),
)
for i in (1, 2, 3, 4)
),
VistapoolSwitchEntityDescription(
key="electrolysis_cover",
translation_key="electrolysis_cover",
value_path="hidro.cover_enabled",
exists_path=PATH_HASHIDRO,
),
VistapoolSwitchEntityDescription(
key="electrolysis_boost",
translation_key="electrolysis_boost",
value_path="hidro.cloration_enabled",
exists_path=PATH_HASHIDRO,
),
VistapoolSwitchEntityDescription(
key="heating_climate",
translation_key="heating_climate",
value_path="filtration.heating.clima",
exists_path="filtration.hasHeat",
),
VistapoolSwitchEntityDescription(
key="smart_mode_freeze",
translation_key="smart_mode_freeze",
value_path="filtration.smart.freeze",
exists_path="filtration.hasSmart",
),
)
def _build_switch_entities(
coordinator: VistapoolDataUpdateCoordinator,
) -> list[SwitchEntity]:
"""Build the switch entities for a single pool."""
entities: list[SwitchEntity] = []
for description in SWITCH_DESCRIPTIONS:
if description.exists_path is not None:
required = (
(description.exists_path,)
if isinstance(description.exists_path, str)
else description.exists_path
)
if not all(coordinator.get_value(path) for path in required):
continue
entities.append(VistapoolSwitch(coordinator, description))
return entities
async def async_setup_entry(
hass: HomeAssistant,
entry: VistapoolConfigEntry,
async_add_entities: AddConfigEntryEntitiesCallback,
) -> None:
"""Set up Vistapool switch entities for every pool on the account."""
entities: list[SwitchEntity] = []
for coordinator in entry.runtime_data.coordinators.values():
entities.extend(_build_switch_entities(coordinator))
async_add_entities(entities)
@callback
def _async_add_pool(coordinator: VistapoolDataUpdateCoordinator) -> None:
async_add_entities(_build_switch_entities(coordinator))
entry.async_on_unload(
async_dispatcher_connect(
hass, f"{SIGNAL_NEW_POOL}_{entry.entry_id}", _async_add_pool
)
)
class VistapoolSwitch(VistapoolEntity, SwitchEntity):
"""Generic Vistapool switch driven by an entity description."""
entity_description: VistapoolSwitchEntityDescription
def __init__(
self,
coordinator: VistapoolDataUpdateCoordinator,
description: VistapoolSwitchEntityDescription,
) -> None:
"""Initialize the switch."""
super().__init__(coordinator)
self.entity_description = description
self._attr_unique_id = self.build_unique_id(description.key)
@property
@override
def is_on(self) -> bool | None:
"""Return true if the writable path or any extra read path is truthy."""
value = self.coordinator.get_value(self.entity_description.value_path)
if value is None:
return None
on = value in (True, "1")
for extra_path in self.entity_description.extra_read_paths:
extra = self.coordinator.get_value(extra_path)
if extra is not None:
on = on or extra in (True, "1")
return on
@override
async def async_turn_on(self, **kwargs: Any) -> None:
"""Turn the switch on."""
await self._async_set_value(1)
@override
async def async_turn_off(self, **kwargs: Any) -> None:
"""Turn the switch off."""
await self._async_set_value(0)
async def _async_set_value(self, value: int) -> None:
"""Send a value update via the Vistapool cloud API."""
try:
await self.coordinator.api.set_value(
self.coordinator.pool_id,
self.entity_description.value_path,
value,
)
except AquariteError as err:
raise HomeAssistantError(
translation_domain=DOMAIN,
translation_key="set_failed",
translation_placeholders={"entity": self.entity_id},
) from err
self.coordinator.apply_optimistic(self.entity_description.value_path, value)
@@ -0,0 +1,351 @@
# serializer version: 1
# name: test_all_entities[switch.my_pool_electrolysis_boost-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_electrolysis_boost',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electrolysis boost',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electrolysis boost',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electrolysis_boost',
'unique_id': 'ABCDEF1234567890-electrolysis_boost',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_electrolysis_boost-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Electrolysis boost',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_electrolysis_boost',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[switch.my_pool_electrolysis_cover-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_electrolysis_cover',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Electrolysis cover',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Electrolysis cover',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'electrolysis_cover',
'unique_id': 'ABCDEF1234567890-electrolysis_cover',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_electrolysis_cover-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Electrolysis cover',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_electrolysis_cover',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[switch.my_pool_filtration-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_filtration',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Filtration',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Filtration',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'filtration',
'unique_id': 'ABCDEF1234567890-filtration',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_filtration-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Filtration',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_filtration',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'on',
})
# ---
# name: test_all_entities[switch.my_pool_relay_1-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_relay_1',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Relay 1',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Relay 1',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay',
'unique_id': 'ABCDEF1234567890-relay_1',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_relay_1-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Relay 1',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_relay_1',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[switch.my_pool_relay_2-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_relay_2',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Relay 2',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Relay 2',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay',
'unique_id': 'ABCDEF1234567890-relay_2',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_relay_2-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Relay 2',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_relay_2',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[switch.my_pool_relay_3-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_relay_3',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Relay 3',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Relay 3',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay',
'unique_id': 'ABCDEF1234567890-relay_3',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_relay_3-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Relay 3',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_relay_3',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
# name: test_all_entities[switch.my_pool_relay_4-entry]
EntityRegistryEntrySnapshot({
'aliases': list([
None,
]),
'area_id': None,
'capabilities': None,
'config_entry_id': <ANY>,
'config_subentry_id': <ANY>,
'device_class': None,
'device_id': <ANY>,
'disabled_by': None,
'domain': 'switch',
'entity_category': None,
'entity_id': 'switch.my_pool_relay_4',
'has_entity_name': True,
'hidden_by': None,
'icon': None,
'id': <ANY>,
'labels': set({
}),
'name': None,
'object_id_base': 'Relay 4',
'options': dict({
}),
'original_device_class': None,
'original_icon': None,
'original_name': 'Relay 4',
'platform': 'vistapool',
'previous_unique_id': None,
'suggested_object_id': None,
'supported_features': 0,
'translation_key': 'relay',
'unique_id': 'ABCDEF1234567890-relay_4',
'unit_of_measurement': None,
})
# ---
# name: test_all_entities[switch.my_pool_relay_4-state]
StateSnapshot({
'attributes': ReadOnlyDict({
<EntityStateAttribute.FRIENDLY_NAME: 'friendly_name'>: 'My Pool Relay 4',
}),
'context': <ANY>,
'entity_id': 'switch.my_pool_relay_4',
'last_changed': <ANY>,
'last_reported': <ANY>,
'last_updated': <ANY>,
'state': 'off',
})
# ---
+167
View File
@@ -0,0 +1,167 @@
"""Tests for the Vistapool switch platform."""
from collections.abc import Generator
from typing import Any
from unittest.mock import AsyncMock, patch
from aioaquarite import AquariteError
import pytest
from syrupy.assertion import SnapshotAssertion
from homeassistant.components.switch import (
DOMAIN as SWITCH_DOMAIN,
SERVICE_TURN_OFF,
SERVICE_TURN_ON,
)
from homeassistant.const import ATTR_ENTITY_ID, STATE_OFF, STATE_ON, Platform
from homeassistant.core import HomeAssistant
from homeassistant.exceptions import HomeAssistantError
from homeassistant.helpers import entity_registry as er
from tests.common import MockConfigEntry, snapshot_platform
@pytest.fixture(autouse=True)
def _only_switch_platform() -> Generator[None]:
"""Restrict integration setup to the switch platform for these tests."""
with patch("homeassistant.components.vistapool.PLATFORMS", [Platform.SWITCH]):
yield
async def test_all_entities(
hass: HomeAssistant,
snapshot: SnapshotAssertion,
entity_registry: er.EntityRegistry,
mock_config_entry: MockConfigEntry,
mock_vistapool_client: AsyncMock,
mock_pool_data: dict[str, Any],
) -> None:
"""Test switch entities for the default fixture."""
mock_vistapool_client.fetch_pool_data.return_value = mock_pool_data
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
async def test_switch_relay_status_fallback(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_vistapool_client: AsyncMock,
) -> None:
"""Test relay switches report on when the read-only status is set."""
mock_vistapool_client.fetch_pool_data.return_value = {
"main": {"version": 1},
"relays": {
"relay1": {"info": {"onoff": 0, "status": 1}},
"relay2": {"info": {"onoff": 1, "status": 0}},
"relay3": {"info": {"onoff": 0, "status": 0}},
"relay4": {"info": {"onoff": 0, "status": 0}},
},
}
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("switch.my_pool_relay_1").state == STATE_ON
assert hass.states.get("switch.my_pool_relay_2").state == STATE_ON
assert hass.states.get("switch.my_pool_relay_3").state == STATE_OFF
async def test_switch_heating_climate_requires_has_heat(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_vistapool_client: AsyncMock,
) -> None:
"""Test heating_climate is only created when filtration.hasHeat is set."""
mock_vistapool_client.fetch_pool_data.return_value = {
"main": {"version": 1},
"filtration": {"hasHeat": 1, "heating": {"clima": 0}},
}
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("switch.my_pool_heating_climate") is not None
assert hass.states.get("switch.my_pool_smart_mode_freeze") is None
async def test_switch_smart_mode_freeze_requires_has_smart(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_vistapool_client: AsyncMock,
) -> None:
"""Test smart_mode_freeze is only created when filtration.hasSmart is set."""
mock_vistapool_client.fetch_pool_data.return_value = {
"main": {"version": 1},
"filtration": {"hasSmart": 1, "smart": {"freeze": 1}},
}
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
assert hass.states.get("switch.my_pool_smart_mode_freeze").state == STATE_ON
assert hass.states.get("switch.my_pool_heating_climate") is None
@pytest.mark.parametrize(
("service", "expected_value"),
[
pytest.param(SERVICE_TURN_ON, 1, id="turn_on"),
pytest.param(SERVICE_TURN_OFF, 0, id="turn_off"),
],
)
async def test_switch_set_value(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_vistapool_client: AsyncMock,
mock_pool_data: dict[str, Any],
service: str,
expected_value: int,
) -> None:
"""Test turn_on / turn_off call the library's set_value with the right args."""
mock_vistapool_client.fetch_pool_data.return_value = mock_pool_data
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
await hass.services.async_call(
SWITCH_DOMAIN,
service,
{ATTR_ENTITY_ID: "switch.my_pool_filtration"},
blocking=True,
)
mock_vistapool_client.set_value.assert_awaited_once_with(
"ABCDEF1234567890", "filtration.status", expected_value
)
async def test_switch_set_value_raises_on_api_error(
hass: HomeAssistant,
mock_config_entry: MockConfigEntry,
mock_vistapool_client: AsyncMock,
mock_pool_data: dict[str, Any],
) -> None:
"""Test action raises HomeAssistantError when the library fails."""
mock_vistapool_client.fetch_pool_data.return_value = mock_pool_data
mock_vistapool_client.set_value.side_effect = AquariteError("boom")
mock_config_entry.add_to_hass(hass)
assert await hass.config_entries.async_setup(mock_config_entry.entry_id)
await hass.async_block_till_done()
with pytest.raises(HomeAssistantError) as excinfo:
await hass.services.async_call(
SWITCH_DOMAIN,
SERVICE_TURN_ON,
{ATTR_ENTITY_ID: "switch.my_pool_filtration"},
blocking=True,
)
assert excinfo.value.translation_key == "set_failed"