mirror of
https://github.com/home-assistant/core.git
synced 2026-09-13 04:01:03 +01:00
Add binary_sensor platform to NeoPool (#180393)
This commit is contained in:
Miloš Svašek
@@ -0,0 +1,390 @@
|
||||
"""Binary sensor platform for the NeoPool integration."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, override
|
||||
|
||||
from neopool_modbus.capabilities import (
|
||||
has_heating_relay,
|
||||
is_chlorine_module_present,
|
||||
is_conductivity_module_present,
|
||||
is_hydrolysis_present,
|
||||
is_ionization_present,
|
||||
is_ph_module_present,
|
||||
is_redox_module_present,
|
||||
)
|
||||
from neopool_modbus.registers import is_valid_relay_gpio
|
||||
|
||||
from homeassistant.components.binary_sensor import (
|
||||
BinarySensorDeviceClass,
|
||||
BinarySensorEntity,
|
||||
BinarySensorEntityDescription,
|
||||
)
|
||||
from homeassistant.const import EntityCategory
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from .const import (
|
||||
CONF_USE_AUX1,
|
||||
CONF_USE_AUX2,
|
||||
CONF_USE_AUX3,
|
||||
CONF_USE_AUX4,
|
||||
CONF_USE_COVER_SENSOR,
|
||||
CONF_USE_LIGHT,
|
||||
)
|
||||
from .coordinator import NeoPoolConfigEntry, NeoPoolCoordinator
|
||||
from .entity import NeoPoolEntity
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
type _SupportedFn = Callable[[dict[str, Any]], bool]
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class NeoPoolBinarySensorEntityDescription(BinarySensorEntityDescription):
|
||||
"""Describes a NeoPool binary sensor entity."""
|
||||
|
||||
supported_fn: _SupportedFn | None = None
|
||||
value_fn: Callable[[dict[str, Any], HomeAssistant], bool | None] | None = None
|
||||
|
||||
|
||||
def _gpio_ok(gpio_key: str) -> _SupportedFn:
|
||||
"""Return a supported_fn that checks a relay GPIO key is valid."""
|
||||
return lambda data: gpio_key not in data or is_valid_relay_gpio(data[gpio_key] or 0)
|
||||
|
||||
|
||||
def _pool_cover_open(data: dict[str, Any], hass: HomeAssistant) -> bool | None:
|
||||
"""Invert the raw cover state for the OPENING device class.
|
||||
|
||||
The cover bit is only valid while filtration runs; otherwise report unknown.
|
||||
"""
|
||||
if data.get("Filtration Pump") is not True:
|
||||
return None
|
||||
value = data.get("Pool Cover")
|
||||
if value is None:
|
||||
return None
|
||||
return not bool(value)
|
||||
|
||||
|
||||
BINARY_SENSOR_DESCRIPTIONS: dict[str, NeoPoolBinarySensorEntityDescription] = {
|
||||
"pH Acid Pump": NeoPoolBinarySensorEntityDescription(
|
||||
key="pH Acid Pump",
|
||||
translation_key="ph_acid_pump",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=_gpio_ok("MBF_PAR_PH_ACID_RELAY_GPIO"),
|
||||
),
|
||||
"Filtration Pump": NeoPoolBinarySensorEntityDescription(
|
||||
key="Filtration Pump",
|
||||
translation_key="filtration_pump",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
supported_fn=_gpio_ok("MBF_PAR_FILT_GPIO"),
|
||||
),
|
||||
"Pool Light": NeoPoolBinarySensorEntityDescription(
|
||||
key="Pool Light",
|
||||
translation_key="pool_light",
|
||||
device_class=BinarySensorDeviceClass.LIGHT,
|
||||
supported_fn=_gpio_ok("MBF_PAR_LIGHTING_GPIO"),
|
||||
),
|
||||
"AUX1": NeoPoolBinarySensorEntityDescription(
|
||||
key="AUX1",
|
||||
translation_key="aux",
|
||||
translation_placeholders={"number": "1"},
|
||||
device_class=BinarySensorDeviceClass.POWER,
|
||||
),
|
||||
"AUX2": NeoPoolBinarySensorEntityDescription(
|
||||
key="AUX2",
|
||||
translation_key="aux",
|
||||
translation_placeholders={"number": "2"},
|
||||
device_class=BinarySensorDeviceClass.POWER,
|
||||
),
|
||||
"AUX3": NeoPoolBinarySensorEntityDescription(
|
||||
key="AUX3",
|
||||
translation_key="aux",
|
||||
translation_placeholders={"number": "3"},
|
||||
device_class=BinarySensorDeviceClass.POWER,
|
||||
),
|
||||
"AUX4": NeoPoolBinarySensorEntityDescription(
|
||||
key="AUX4",
|
||||
translation_key="aux",
|
||||
translation_placeholders={"number": "4"},
|
||||
device_class=BinarySensorDeviceClass.POWER,
|
||||
),
|
||||
"pH module control status": NeoPoolBinarySensorEntityDescription(
|
||||
key="pH module control status",
|
||||
translation_key="ph_module_control_status",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_ph_module_present,
|
||||
),
|
||||
"pH control module": NeoPoolBinarySensorEntityDescription(
|
||||
key="pH control module",
|
||||
translation_key="ph_control_module",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_ph_module_present,
|
||||
),
|
||||
"pH measurement active": NeoPoolBinarySensorEntityDescription(
|
||||
key="pH measurement active",
|
||||
translation_key="ph_measurement_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_ph_module_present,
|
||||
),
|
||||
"Redox pump active": NeoPoolBinarySensorEntityDescription(
|
||||
key="Redox pump active",
|
||||
translation_key="redox_pump_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=lambda data: (
|
||||
is_redox_module_present(data)
|
||||
and (
|
||||
"MBF_PAR_RX_RELAY_GPIO" not in data
|
||||
or is_valid_relay_gpio(data["MBF_PAR_RX_RELAY_GPIO"] or 0)
|
||||
)
|
||||
),
|
||||
),
|
||||
"Redox control module": NeoPoolBinarySensorEntityDescription(
|
||||
key="Redox control module",
|
||||
translation_key="redox_control_module",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_redox_module_present,
|
||||
),
|
||||
"Redox measurement active": NeoPoolBinarySensorEntityDescription(
|
||||
key="Redox measurement active",
|
||||
translation_key="redox_measurement_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_redox_module_present,
|
||||
),
|
||||
"Chlorine flow sensor problem": NeoPoolBinarySensorEntityDescription(
|
||||
key="Chlorine flow sensor problem",
|
||||
translation_key="chlorine_flow_sensor_problem",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=is_chlorine_module_present,
|
||||
),
|
||||
"Chlorine pump active": NeoPoolBinarySensorEntityDescription(
|
||||
key="Chlorine pump active",
|
||||
translation_key="chlorine_pump_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=lambda data: (
|
||||
is_chlorine_module_present(data)
|
||||
and (
|
||||
"MBF_PAR_CL_RELAY_GPIO" not in data
|
||||
or is_valid_relay_gpio(data["MBF_PAR_CL_RELAY_GPIO"] or 0)
|
||||
)
|
||||
),
|
||||
),
|
||||
"Chlorine control module": NeoPoolBinarySensorEntityDescription(
|
||||
key="Chlorine control module",
|
||||
translation_key="chlorine_control_module",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_chlorine_module_present,
|
||||
),
|
||||
"Chlorine measurement active": NeoPoolBinarySensorEntityDescription(
|
||||
key="Chlorine measurement active",
|
||||
translation_key="chlorine_measurement_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_chlorine_module_present,
|
||||
),
|
||||
"Conductivity pump active": NeoPoolBinarySensorEntityDescription(
|
||||
key="Conductivity pump active",
|
||||
translation_key="conductivity_pump_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=lambda data: (
|
||||
is_conductivity_module_present(data)
|
||||
and (
|
||||
"MBF_PAR_CD_RELAY_GPIO" not in data
|
||||
or is_valid_relay_gpio(data["MBF_PAR_CD_RELAY_GPIO"] or 0)
|
||||
)
|
||||
),
|
||||
),
|
||||
"Conductivity control module": NeoPoolBinarySensorEntityDescription(
|
||||
key="Conductivity control module",
|
||||
translation_key="conductivity_control_module",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_conductivity_module_present,
|
||||
),
|
||||
"Conductivity measurement active": NeoPoolBinarySensorEntityDescription(
|
||||
key="Conductivity measurement active",
|
||||
translation_key="conductivity_measurement_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_conductivity_module_present,
|
||||
),
|
||||
"ION On Target": NeoPoolBinarySensorEntityDescription(
|
||||
key="ION On Target",
|
||||
translation_key="ion_on_target",
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_ionization_present, # pragma: no cover
|
||||
),
|
||||
"ION Low": NeoPoolBinarySensorEntityDescription(
|
||||
key="ION Low",
|
||||
translation_key="ion_low",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=is_ionization_present, # pragma: no cover
|
||||
),
|
||||
"ION Program time exceeded": NeoPoolBinarySensorEntityDescription(
|
||||
key="ION Program time exceeded",
|
||||
translation_key="ion_program_time_exceeded",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=is_ionization_present, # pragma: no cover
|
||||
),
|
||||
"HIDRO Low": NeoPoolBinarySensorEntityDescription(
|
||||
key="HIDRO Low",
|
||||
translation_key="hidro_low",
|
||||
device_class=BinarySensorDeviceClass.PROBLEM,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"Pool Cover": NeoPoolBinarySensorEntityDescription(
|
||||
key="Pool Cover",
|
||||
translation_key="pool_cover",
|
||||
device_class=BinarySensorDeviceClass.OPENING,
|
||||
value_fn=_pool_cover_open,
|
||||
),
|
||||
"HIDRO Module active": NeoPoolBinarySensorEntityDescription(
|
||||
key="HIDRO Module active",
|
||||
translation_key="hidro_module_active",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"HIDRO Module regulated": NeoPoolBinarySensorEntityDescription(
|
||||
key="HIDRO Module regulated",
|
||||
translation_key="hidro_module_regulated",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"HIDRO Activated by the RX module": NeoPoolBinarySensorEntityDescription(
|
||||
key="HIDRO Activated by the RX module",
|
||||
translation_key="hidro_activated_by_the_rx_module",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=lambda data: (
|
||||
is_hydrolysis_present(data) and is_redox_module_present(data)
|
||||
), # pragma: no cover
|
||||
),
|
||||
"HIDRO Chlorine shock mode": NeoPoolBinarySensorEntityDescription(
|
||||
key="HIDRO Chlorine shock mode",
|
||||
translation_key="hidro_chlorine_shock_mode",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"HIDRO Activated by the CL module": NeoPoolBinarySensorEntityDescription(
|
||||
key="HIDRO Activated by the CL module",
|
||||
translation_key="hidro_activated_by_the_cl_module",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=lambda data: (
|
||||
is_hydrolysis_present(data) and is_chlorine_module_present(data)
|
||||
),
|
||||
),
|
||||
"Heating": NeoPoolBinarySensorEntityDescription(
|
||||
key="Heating",
|
||||
translation_key="heating",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=has_heating_relay,
|
||||
),
|
||||
"UV Lamp": NeoPoolBinarySensorEntityDescription(
|
||||
key="UV Lamp",
|
||||
translation_key="uv_lamp",
|
||||
device_class=BinarySensorDeviceClass.RUNNING,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=lambda data: (
|
||||
"MBF_PAR_UV_RELAY_GPIO" not in data
|
||||
or is_valid_relay_gpio(data["MBF_PAR_UV_RELAY_GPIO"] or 0)
|
||||
),
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
# Entities gated on a config-entry option (in addition to their supported_fn).
|
||||
# The controller cannot detect what is physically wired to the light or aux
|
||||
# relays, nor whether a cover sensor is present, so these entities are opt-in
|
||||
# per config entry rather than surfaced from a device capability bit.
|
||||
_ENTITY_OPTION_KEY: dict[str, str] = {
|
||||
"Pool Light": CONF_USE_LIGHT,
|
||||
"AUX1": CONF_USE_AUX1,
|
||||
"AUX2": CONF_USE_AUX2,
|
||||
"AUX3": CONF_USE_AUX3,
|
||||
"AUX4": CONF_USE_AUX4,
|
||||
"Pool Cover": CONF_USE_COVER_SENSOR,
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: NeoPoolConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up NeoPool binary sensors from a config entry."""
|
||||
coordinator = entry.runtime_data
|
||||
options = entry.options
|
||||
|
||||
async_add_entities(
|
||||
NeoPoolBinarySensor(coordinator, key, desc)
|
||||
for key, desc in BINARY_SENSOR_DESCRIPTIONS.items()
|
||||
if (
|
||||
(option_key := _ENTITY_OPTION_KEY.get(key)) is None
|
||||
or bool(options.get(option_key))
|
||||
)
|
||||
and (desc.supported_fn is None or desc.supported_fn(coordinator.data))
|
||||
)
|
||||
|
||||
|
||||
class NeoPoolBinarySensor(NeoPoolEntity, BinarySensorEntity):
|
||||
"""Representation of a NeoPool binary sensor."""
|
||||
|
||||
_winter_mode_active = False
|
||||
entity_description: NeoPoolBinarySensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: NeoPoolCoordinator,
|
||||
key: str,
|
||||
description: NeoPoolBinarySensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the binary sensor."""
|
||||
super().__init__(coordinator)
|
||||
self.entity_description = description
|
||||
self._key = key
|
||||
self._attr_unique_id = (
|
||||
f"{self.coordinator.config_entry.unique_id}_{key.lower()}"
|
||||
)
|
||||
|
||||
@property
|
||||
@override
|
||||
def is_on(self) -> bool | None:
|
||||
"""Return True if the binary sensor is on."""
|
||||
if (value_fn := self.entity_description.value_fn) is not None:
|
||||
value: bool | None = value_fn(self.coordinator.data, self.hass)
|
||||
return value
|
||||
value = self.coordinator.data.get(self._key)
|
||||
return None if value is None else bool(value)
|
||||
@@ -6,6 +6,7 @@ DOMAIN = "neopool"
|
||||
NAME = "NeoPool"
|
||||
|
||||
PLATFORMS: list[Platform] = [
|
||||
Platform.BINARY_SENSOR,
|
||||
Platform.BUTTON,
|
||||
Platform.LIGHT,
|
||||
Platform.SENSOR,
|
||||
|
||||
@@ -44,6 +44,95 @@
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"binary_sensor": {
|
||||
"aux": {
|
||||
"name": "Auxiliary relay {number}"
|
||||
},
|
||||
"chlorine_control_module": {
|
||||
"name": "Chlorine regulation active"
|
||||
},
|
||||
"chlorine_flow_sensor_problem": {
|
||||
"name": "Chlorine flow sensor"
|
||||
},
|
||||
"chlorine_measurement_active": {
|
||||
"name": "Chlorine measurement"
|
||||
},
|
||||
"chlorine_pump_active": {
|
||||
"name": "Chlorine pump active"
|
||||
},
|
||||
"conductivity_control_module": {
|
||||
"name": "Conductivity regulation active"
|
||||
},
|
||||
"conductivity_measurement_active": {
|
||||
"name": "Conductivity measurement"
|
||||
},
|
||||
"conductivity_pump_active": {
|
||||
"name": "Conductivity pump active"
|
||||
},
|
||||
"filtration_pump": {
|
||||
"name": "Filtration"
|
||||
},
|
||||
"heating": {
|
||||
"name": "Heating"
|
||||
},
|
||||
"hidro_activated_by_the_cl_module": {
|
||||
"name": "Hydrolysis activated by chlorine module"
|
||||
},
|
||||
"hidro_activated_by_the_rx_module": {
|
||||
"name": "Hydrolysis activated by Redox module"
|
||||
},
|
||||
"hidro_chlorine_shock_mode": {
|
||||
"name": "Hydrolysis chlorine shock mode (boost)"
|
||||
},
|
||||
"hidro_low": {
|
||||
"name": "Hydrolysis production problem"
|
||||
},
|
||||
"hidro_module_active": {
|
||||
"name": "Hydrolysis enabled"
|
||||
},
|
||||
"hidro_module_regulated": {
|
||||
"name": "Hydrolysis regulation active"
|
||||
},
|
||||
"ion_low": {
|
||||
"name": "Ionizer production problem"
|
||||
},
|
||||
"ion_on_target": {
|
||||
"name": "Ionizer on target"
|
||||
},
|
||||
"ion_program_time_exceeded": {
|
||||
"name": "Ionizer program time exceeded"
|
||||
},
|
||||
"ph_acid_pump": {
|
||||
"name": "pH acid pump"
|
||||
},
|
||||
"ph_control_module": {
|
||||
"name": "pH regulation active"
|
||||
},
|
||||
"ph_measurement_active": {
|
||||
"name": "pH measurement"
|
||||
},
|
||||
"ph_module_control_status": {
|
||||
"name": "pH flow detection control"
|
||||
},
|
||||
"pool_cover": {
|
||||
"name": "Pool cover"
|
||||
},
|
||||
"pool_light": {
|
||||
"name": "Pool light"
|
||||
},
|
||||
"redox_control_module": {
|
||||
"name": "Redox regulation active"
|
||||
},
|
||||
"redox_measurement_active": {
|
||||
"name": "Redox measurement"
|
||||
},
|
||||
"redox_pump_active": {
|
||||
"name": "Redox pump active"
|
||||
},
|
||||
"uv_lamp": {
|
||||
"name": "UV lamp"
|
||||
}
|
||||
},
|
||||
"button": {
|
||||
"escape": {
|
||||
"name": "Clear error messages"
|
||||
|
||||
@@ -83,6 +83,15 @@ MOCK_POOL_DATA: dict[str, Any] = {
|
||||
"pH pump active": False,
|
||||
"pH acid pump active": False,
|
||||
"Filtration Pump": False,
|
||||
"pH Acid Pump": False,
|
||||
# Measurement / module "active" bits. The controller keeps measuring the
|
||||
# probes regardless of filtration state, so these read True even though the
|
||||
# filtration pump above is off.
|
||||
"pH measurement active": True,
|
||||
"Redox measurement active": True,
|
||||
"Chlorine measurement active": True,
|
||||
"Conductivity measurement active": True,
|
||||
"HIDRO Module active": True,
|
||||
"MBF_PAR_HIDRO_COVER_REDUCTION": 0x0C19,
|
||||
"MBF_PAR_HIDRO_COVER_ENABLE": 0x0000,
|
||||
"Pool Cover": 0,
|
||||
@@ -200,6 +209,58 @@ def mock_config_entry_switch() -> MockConfigEntry:
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry_binary_sensor() -> MockConfigEntry:
|
||||
"""Return a config entry with the options the binary_sensor platform gates on."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title=MOCK_NAME,
|
||||
unique_id=MOCK_SERIAL,
|
||||
version=CURRENT_VERSION,
|
||||
data={
|
||||
CONF_HOST: MOCK_HOST,
|
||||
CONF_PORT: MOCK_PORT,
|
||||
CONF_NAME: MOCK_NAME,
|
||||
"unit_id": DEFAULT_UNIT_ID,
|
||||
"modbus_framer": "tcp",
|
||||
},
|
||||
options={
|
||||
CONF_USE_LIGHT: True,
|
||||
CONF_USE_COVER_SENSOR: True,
|
||||
CONF_USE_AUX1: True,
|
||||
CONF_USE_AUX2: True,
|
||||
CONF_USE_AUX3: True,
|
||||
CONF_USE_AUX4: True,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry_binary_sensor_no_options() -> MockConfigEntry:
|
||||
"""Return a config entry with every binary_sensor option disabled."""
|
||||
return MockConfigEntry(
|
||||
domain=DOMAIN,
|
||||
title=MOCK_NAME,
|
||||
unique_id=MOCK_SERIAL,
|
||||
version=CURRENT_VERSION,
|
||||
data={
|
||||
CONF_HOST: MOCK_HOST,
|
||||
CONF_PORT: MOCK_PORT,
|
||||
CONF_NAME: MOCK_NAME,
|
||||
"unit_id": DEFAULT_UNIT_ID,
|
||||
"modbus_framer": "tcp",
|
||||
},
|
||||
options={
|
||||
CONF_USE_LIGHT: False,
|
||||
CONF_USE_COVER_SENSOR: False,
|
||||
CONF_USE_AUX1: False,
|
||||
CONF_USE_AUX2: False,
|
||||
CONF_USE_AUX3: False,
|
||||
CONF_USE_AUX4: False,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_neopool_client() -> Generator[MagicMock]:
|
||||
"""Patch the NeoPoolModbusClient and return a configurable mock instance."""
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -31,9 +31,12 @@
|
||||
'CELL_RUNTIME_POLB': 1800,
|
||||
'CELL_RUNTIME_POL_CHANGES': 7,
|
||||
'CELL_RUNTIME_TOTAL': 65536,
|
||||
'Chlorine measurement active': True,
|
||||
'Chlorine measurement module detected': True,
|
||||
'Conductivity measurement active': True,
|
||||
'Conductivity measurement module detected': True,
|
||||
'Filtration Pump': False,
|
||||
'HIDRO Module active': True,
|
||||
'HIDRO in Pol1': False,
|
||||
'HIDRO in Pol2': False,
|
||||
'HIDRO in dead time': False,
|
||||
@@ -81,11 +84,14 @@
|
||||
'MBF_POWER_MODULE_VERSION': 4660,
|
||||
'PH_PUMP_STATUS': 'off',
|
||||
'Pool Cover': 0,
|
||||
'Redox measurement active': True,
|
||||
'Redox measurement module detected': True,
|
||||
'filtration_mode': 'manual',
|
||||
'filtration_speed_state': 'off',
|
||||
'pH Acid Pump': False,
|
||||
'pH acid pump active': False,
|
||||
'pH control module': True,
|
||||
'pH measurement active': True,
|
||||
'pH measurement module detected': True,
|
||||
'pH pump active': False,
|
||||
}),
|
||||
|
||||
@@ -0,0 +1,255 @@
|
||||
"""Tests for the NeoPool binary_sensor platform value decoders."""
|
||||
|
||||
from datetime import timedelta
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
import pytest
|
||||
from syrupy.assertion import SnapshotAssertion
|
||||
|
||||
from homeassistant.components.binary_sensor import DOMAIN as BINARY_SENSOR_DOMAIN
|
||||
from homeassistant.const import STATE_OFF, STATE_ON, STATE_UNKNOWN, Platform
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import entity_registry as er
|
||||
|
||||
from . import setup_integration
|
||||
from .conftest import MOCK_POOL_DATA
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed, snapshot_platform
|
||||
|
||||
|
||||
def _binary_state(hass: HomeAssistant, entry: MockConfigEntry, key: str):
|
||||
"""Return the HA state object of the binary_sensor for a coordinator key."""
|
||||
registry = er.async_get(hass)
|
||||
suffix = f"_{key.lower()}"
|
||||
entries = [
|
||||
e
|
||||
for e in er.async_entries_for_config_entry(registry, entry.entry_id)
|
||||
if e.domain == BINARY_SENSOR_DOMAIN and e.unique_id.endswith(suffix)
|
||||
]
|
||||
if not entries:
|
||||
return None
|
||||
return hass.states.get(entries[0].entity_id)
|
||||
|
||||
|
||||
async def test_direct_key_reflects_coordinator_value(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""A simple boolean key from coordinator.data flows straight through is_on."""
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"Filtration Pump": True,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(hass, mock_config_entry_binary_sensor, "Filtration Pump")
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"Filtration Pump": False,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(hass, mock_config_entry_binary_sensor, "Filtration Pump")
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
async def test_pool_cover_inverts_hardware_value(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Pool Cover: hardware 1 (covered) → HA OFF; hardware 0 → HA ON.
|
||||
|
||||
The OPENING device class needs the opposite polarity from the raw
|
||||
register, so the entity inverts the value before returning is_on.
|
||||
"""
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"Pool Cover": True,
|
||||
"Filtration Pump": True,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(hass, mock_config_entry_binary_sensor, "Pool Cover")
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"Pool Cover": False,
|
||||
"Filtration Pump": True,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(hass, mock_config_entry_binary_sensor, "Pool Cover")
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
|
||||
|
||||
async def test_pool_cover_none_yields_unknown(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Missing Pool Cover key surfaces as STATE_UNKNOWN, not on/off."""
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"Pool Cover": None,
|
||||
"Filtration Pump": True,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(hass, mock_config_entry_binary_sensor, "Pool Cover")
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNKNOWN
|
||||
|
||||
|
||||
@pytest.mark.parametrize("pump_state", [False, None])
|
||||
async def test_pool_cover_unknown_when_filtration_not_running(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
pump_state: bool | None,
|
||||
) -> None:
|
||||
"""Cover reads unknown unless the pump is confirmed running.
|
||||
|
||||
The device only reports the cover bit while filtration runs, so an idle
|
||||
(False) or unknown (None) pump state must not surface a stale open/closed.
|
||||
"""
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"Pool Cover": False,
|
||||
"Filtration Pump": pump_state,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(hass, mock_config_entry_binary_sensor, "Pool Cover")
|
||||
assert state is not None
|
||||
assert state.state == STATE_UNKNOWN
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_measurement_module_reads_raw_bit(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Measurement-module sensors report the raw device bit, even with filtration off.
|
||||
|
||||
The controller keeps measuring the probes regardless of the filtration
|
||||
pump state, so the entity must not force the value off.
|
||||
"""
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"pH measurement active": True,
|
||||
"Filtration Pump": False,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(
|
||||
hass, mock_config_entry_binary_sensor, "pH measurement active"
|
||||
)
|
||||
assert state is not None
|
||||
assert state.state == STATE_ON
|
||||
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"pH measurement active": False,
|
||||
"Filtration Pump": False,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done(wait_background_tasks=True)
|
||||
state = _binary_state(
|
||||
hass, mock_config_entry_binary_sensor, "pH measurement active"
|
||||
)
|
||||
assert state is not None
|
||||
assert state.state == STATE_OFF
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_neopool_client")
|
||||
async def test_all_entities(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
entity_registry: er.EntityRegistry,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Snapshot every entity registered by the binary_sensor platform."""
|
||||
with patch("homeassistant.components.neopool.PLATFORMS", [Platform.BINARY_SENSOR]):
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
await snapshot_platform(
|
||||
hass, entity_registry, snapshot, mock_config_entry_binary_sensor.entry_id
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default")
|
||||
async def test_setup_when_modules_absent(
|
||||
hass: HomeAssistant,
|
||||
snapshot: SnapshotAssertion,
|
||||
entity_registry: er.EntityRegistry,
|
||||
mock_config_entry_binary_sensor: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
minimal_pool_data: dict[str, Any],
|
||||
) -> None:
|
||||
"""Snapshot the binary_sensor entities registered when no modules are present."""
|
||||
mock_neopool_client.async_read_all.return_value = minimal_pool_data
|
||||
with patch("homeassistant.components.neopool.PLATFORMS", [Platform.BINARY_SENSOR]):
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor)
|
||||
await snapshot_platform(
|
||||
hass, entity_registry, snapshot, mock_config_entry_binary_sensor.entry_id
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_neopool_client")
|
||||
async def test_opt_in_entities_absent_without_options(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry_binary_sensor_no_options: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Opt-in entities are not registered when their config option is off.
|
||||
|
||||
Pool Light, the four auxiliary relays, and Pool Cover are gated on an
|
||||
integration option in addition to any capability check. With every option
|
||||
disabled they must not register, while an ungated relay sensor still does.
|
||||
"""
|
||||
with patch("homeassistant.components.neopool.PLATFORMS", [Platform.BINARY_SENSOR]):
|
||||
await setup_integration(hass, mock_config_entry_binary_sensor_no_options)
|
||||
|
||||
for key in ("Pool Light", "AUX1", "AUX2", "AUX3", "AUX4", "Pool Cover"):
|
||||
assert (
|
||||
_binary_state(hass, mock_config_entry_binary_sensor_no_options, key) is None
|
||||
)
|
||||
assert (
|
||||
_binary_state(
|
||||
hass, mock_config_entry_binary_sensor_no_options, "Filtration Pump"
|
||||
)
|
||||
is not None
|
||||
)
|
||||
Reference in New Issue
Block a user