mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 10:13:52 -05:00
Add NeoPool integration (#173779)
This commit is contained in:
Miloš Svašek
@@ -395,6 +395,7 @@ homeassistant.components.nam.*
|
||||
homeassistant.components.namecheapdns.*
|
||||
homeassistant.components.nasweb.*
|
||||
homeassistant.components.neato.*
|
||||
homeassistant.components.neopool.*
|
||||
homeassistant.components.nest.*
|
||||
homeassistant.components.netatmo.*
|
||||
homeassistant.components.network.*
|
||||
|
||||
Generated
+2
@@ -1193,6 +1193,8 @@ CLAUDE.md @home-assistant/core
|
||||
/tests/components/nasweb/ @nasWebio
|
||||
/homeassistant/components/nederlandse_spoorwegen/ @YarmoM @heindrichpaul
|
||||
/tests/components/nederlandse_spoorwegen/ @YarmoM @heindrichpaul
|
||||
/homeassistant/components/neopool/ @svasek
|
||||
/tests/components/neopool/ @svasek
|
||||
/homeassistant/components/ness_alarm/ @nickw444 @poshy163
|
||||
/tests/components/ness_alarm/ @nickw444 @poshy163
|
||||
/homeassistant/components/nest/ @allenporter
|
||||
|
||||
@@ -0,0 +1,28 @@
|
||||
"""NeoPool integration for Home Assistant."""
|
||||
|
||||
from neopool_modbus import NeoPoolModbusClient
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from .const import PLATFORMS
|
||||
from .coordinator import NeoPoolConfigEntry, NeoPoolCoordinator
|
||||
|
||||
|
||||
async def async_setup_entry(hass: HomeAssistant, entry: NeoPoolConfigEntry) -> bool:
|
||||
"""Set up the NeoPool integration from a config entry."""
|
||||
client = NeoPoolModbusClient(entry.data)
|
||||
coordinator = NeoPoolCoordinator(hass, client, entry)
|
||||
await coordinator.async_config_entry_first_refresh()
|
||||
entry.runtime_data = coordinator
|
||||
|
||||
await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
async def async_unload_entry(hass: HomeAssistant, entry: NeoPoolConfigEntry) -> bool:
|
||||
"""Unload a NeoPool config entry."""
|
||||
unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
|
||||
if unload_ok:
|
||||
await entry.runtime_data.client.close()
|
||||
return unload_ok
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Config flow for the NeoPool integration."""
|
||||
|
||||
from typing import Any, override
|
||||
|
||||
from neopool_modbus import async_probe_serial
|
||||
from neopool_modbus.exceptions import (
|
||||
NeoPoolConnectionError,
|
||||
NeoPoolModbusError,
|
||||
NeoPoolTimeoutError,
|
||||
)
|
||||
from neopool_modbus.registers import DEFAULT_MODBUS_FRAMER
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.config_entries import ConfigFlow, ConfigFlowResult
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
|
||||
from .const import CURRENT_VERSION, DEFAULT_PORT, DEFAULT_UNIT_ID, DOMAIN
|
||||
|
||||
|
||||
async def _async_probe(user_input: dict[str, Any]) -> tuple[str | None, str | None]:
|
||||
"""Probe a device using user-supplied connection parameters."""
|
||||
try:
|
||||
serial = await async_probe_serial(
|
||||
user_input[CONF_HOST],
|
||||
port=user_input[CONF_PORT],
|
||||
unit_id=user_input["unit_id"],
|
||||
framer=user_input["modbus_framer"],
|
||||
)
|
||||
except NeoPoolConnectionError, NeoPoolTimeoutError:
|
||||
return None, "cannot_connect"
|
||||
except NeoPoolModbusError:
|
||||
return None, "cannot_read_modbus"
|
||||
return serial, None
|
||||
|
||||
|
||||
class NeoPoolConfigFlow(ConfigFlow, domain=DOMAIN):
|
||||
"""Handle a config flow for NeoPool."""
|
||||
|
||||
VERSION = CURRENT_VERSION
|
||||
|
||||
@override
|
||||
async def async_step_user(
|
||||
self, user_input: dict[str, Any] | None = None
|
||||
) -> ConfigFlowResult:
|
||||
"""Handle the initial step of the configuration flow."""
|
||||
data_schema = vol.Schema(
|
||||
{
|
||||
vol.Required(CONF_HOST): str,
|
||||
vol.Optional(CONF_PORT, default=DEFAULT_PORT): vol.Coerce(int),
|
||||
vol.Optional("unit_id", default=DEFAULT_UNIT_ID): vol.Coerce(int),
|
||||
vol.Optional(
|
||||
"modbus_framer",
|
||||
default=DEFAULT_MODBUS_FRAMER,
|
||||
): vol.In(("tcp", "rtu")),
|
||||
}
|
||||
)
|
||||
errors: dict[str, str] = {}
|
||||
if user_input is not None:
|
||||
serial, error_key = await _async_probe(user_input)
|
||||
if error_key:
|
||||
errors[CONF_HOST] = error_key
|
||||
else:
|
||||
assert serial is not None
|
||||
await self.async_set_unique_id(serial)
|
||||
self._abort_if_unique_id_configured()
|
||||
|
||||
return self.async_create_entry(
|
||||
title=user_input[CONF_HOST], data=user_input
|
||||
)
|
||||
|
||||
return self.async_show_form(
|
||||
step_id="user",
|
||||
data_schema=data_schema,
|
||||
errors=errors,
|
||||
)
|
||||
@@ -0,0 +1,14 @@
|
||||
"""Constants for the NeoPool integration."""
|
||||
|
||||
from homeassistant.const import Platform
|
||||
|
||||
DOMAIN = "neopool"
|
||||
NAME = "NeoPool"
|
||||
|
||||
PLATFORMS: list[Platform] = [Platform.SENSOR]
|
||||
|
||||
DEFAULT_SCAN_INTERVAL = 20 # in seconds
|
||||
DEFAULT_PORT = 502
|
||||
DEFAULT_UNIT_ID = 1
|
||||
|
||||
CURRENT_VERSION = 6
|
||||
@@ -0,0 +1,100 @@
|
||||
"""Data update coordinator for the NeoPool integration."""
|
||||
|
||||
from datetime import timedelta
|
||||
import logging
|
||||
from typing import Any, override
|
||||
|
||||
from neopool_modbus import NeoPoolModbusClient
|
||||
from neopool_modbus.exceptions import NeoPoolError
|
||||
from neopool_modbus.registers import MAX_RELAY_GPIO, find_corrupted_gpio_registers
|
||||
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import issue_registry as ir
|
||||
from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed
|
||||
|
||||
from .const import DEFAULT_SCAN_INTERVAL, DOMAIN
|
||||
|
||||
_LOGGER = logging.getLogger(__name__)
|
||||
|
||||
|
||||
type NeoPoolConfigEntry = ConfigEntry["NeoPoolCoordinator"]
|
||||
|
||||
|
||||
class NeoPoolCoordinator(DataUpdateCoordinator[dict[str, Any]]):
|
||||
"""Coordinator for NeoPool platform."""
|
||||
|
||||
client: NeoPoolModbusClient
|
||||
config_entry: NeoPoolConfigEntry
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
hass: HomeAssistant,
|
||||
client: NeoPoolModbusClient,
|
||||
entry: NeoPoolConfigEntry,
|
||||
) -> None:
|
||||
"""Initialise the NeoPool data update coordinator."""
|
||||
super().__init__(
|
||||
hass,
|
||||
_LOGGER,
|
||||
name=f"{DOMAIN} coordinator",
|
||||
update_interval=timedelta(seconds=DEFAULT_SCAN_INTERVAL),
|
||||
config_entry=entry,
|
||||
)
|
||||
self.client = client
|
||||
self._corrupted_gpio_state: frozenset[tuple[str, int]] | None = None
|
||||
|
||||
def _check_gpio_registers(self, data: dict[str, Any]) -> None:
|
||||
"""Validate GPIO register values and (re-)raise or clear the repair issue."""
|
||||
corrupted = find_corrupted_gpio_registers(data)
|
||||
corrupted_state = frozenset((key, value) for key, _, value in corrupted)
|
||||
|
||||
if corrupted_state == self._corrupted_gpio_state:
|
||||
return
|
||||
|
||||
for key, label, value in corrupted:
|
||||
_LOGGER.error(
|
||||
"Corrupted GPIO register %s (%s): value %d (0x%04X) is outside "
|
||||
"valid range 0-%d. The pool controller may malfunction",
|
||||
key,
|
||||
label,
|
||||
value,
|
||||
value & 0xFFFF,
|
||||
MAX_RELAY_GPIO,
|
||||
)
|
||||
|
||||
self._corrupted_gpio_state = corrupted_state
|
||||
|
||||
if corrupted:
|
||||
details = "\n".join(
|
||||
f"- **{label}** (`{key}`): value **{value}** (expected 0-{MAX_RELAY_GPIO})"
|
||||
for key, label, value in corrupted
|
||||
)
|
||||
ir.async_create_issue(
|
||||
self.hass,
|
||||
DOMAIN,
|
||||
"corrupted_gpio",
|
||||
is_fixable=False,
|
||||
severity=ir.IssueSeverity.ERROR,
|
||||
translation_key="corrupted_gpio",
|
||||
translation_placeholders={"details": details},
|
||||
)
|
||||
else:
|
||||
# Clear a previously raised repair issue once the device is healthy.
|
||||
ir.async_delete_issue(self.hass, DOMAIN, "corrupted_gpio")
|
||||
|
||||
@override
|
||||
async def _async_update_data(self) -> dict[str, Any]:
|
||||
"""Fetch the latest data from the pool controller."""
|
||||
try:
|
||||
data = await self.client.async_read_all()
|
||||
except (NeoPoolError, OSError, TimeoutError) as err:
|
||||
raise UpdateFailed(
|
||||
translation_domain=DOMAIN,
|
||||
translation_key="modbus_communication_error",
|
||||
translation_placeholders={"error": str(err)},
|
||||
) from err
|
||||
|
||||
self._check_gpio_registers(data)
|
||||
|
||||
return data
|
||||
@@ -0,0 +1,36 @@
|
||||
"""Base entity class for the NeoPool integration."""
|
||||
|
||||
from typing import override
|
||||
|
||||
from neopool_modbus.decoders import get_machine_name, parse_version
|
||||
|
||||
from homeassistant.helpers.device_registry import DeviceInfo
|
||||
from homeassistant.helpers.update_coordinator import CoordinatorEntity
|
||||
|
||||
from .const import DOMAIN, NAME
|
||||
from .coordinator import NeoPoolCoordinator
|
||||
|
||||
|
||||
class NeoPoolEntity(CoordinatorEntity[NeoPoolCoordinator]):
|
||||
"""Base class for NeoPool entities."""
|
||||
|
||||
_attr_has_entity_name = True
|
||||
|
||||
@property
|
||||
@override
|
||||
def device_info(self) -> DeviceInfo:
|
||||
"""Return device information for the entity."""
|
||||
data = self.coordinator.data or {}
|
||||
unique_id = self.coordinator.config_entry.unique_id
|
||||
assert unique_id is not None
|
||||
machine_type = (get_machine_name(data) or "").strip()
|
||||
model_prefix = "NeoPool Compatible: " if machine_type else "NeoPool Compatible"
|
||||
|
||||
return DeviceInfo(
|
||||
identifiers={(DOMAIN, unique_id)},
|
||||
name=NAME,
|
||||
model=f"{model_prefix}{machine_type}".strip(),
|
||||
manufacturer="Hayward (Sugar Valley)",
|
||||
sw_version=f"v{parse_version(data.get('MBF_POWER_MODULE_VERSION'))} (v{parse_version(data.get('MBF_PAR_VERSION'))})",
|
||||
serial_number=unique_id,
|
||||
)
|
||||
@@ -0,0 +1,65 @@
|
||||
{
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"filt_mode": {
|
||||
"default": "mdi:water-sync",
|
||||
"state": {
|
||||
"auto": "mdi:water-boiler-auto",
|
||||
"backwash": "mdi:water-boiler-off",
|
||||
"heating": "mdi:water-boiler-alert",
|
||||
"intelligent": "mdi:water-boiler-auto",
|
||||
"manual": "mdi:water-boiler-alert",
|
||||
"smart": "mdi:water-boiler-auto"
|
||||
}
|
||||
},
|
||||
"filtration_speed": {
|
||||
"default": "mdi:fan"
|
||||
},
|
||||
"filtvalve_remaining": {
|
||||
"default": "mdi:timer-sand"
|
||||
},
|
||||
"hidro_current": {
|
||||
"default": "mdi:air-humidifier-off",
|
||||
"range": {
|
||||
"10": "mdi:air-humidifier"
|
||||
}
|
||||
},
|
||||
"hidro_polarity": {
|
||||
"default": "mdi:plus-minus-variant"
|
||||
},
|
||||
"intelligent_intervals": {
|
||||
"default": "mdi:counter"
|
||||
},
|
||||
"intelligent_tt_next_interval": {
|
||||
"default": "mdi:timeline-clock-outline"
|
||||
},
|
||||
"ion_current": {
|
||||
"default": "mdi:atom"
|
||||
},
|
||||
"ion_polarity": {
|
||||
"default": "mdi:plus-minus-variant"
|
||||
},
|
||||
"measure_cl": {
|
||||
"default": "mdi:shaker-outline"
|
||||
},
|
||||
"measure_rx": {
|
||||
"default": "mdi:gradient-vertical"
|
||||
},
|
||||
"ph_pump_status": {
|
||||
"default": "mdi:pump"
|
||||
},
|
||||
"ph_status_alarm": {
|
||||
"default": "mdi:ph",
|
||||
"state": {
|
||||
"ok": "mdi:check-circle-outline",
|
||||
"ph_high": "mdi:alert",
|
||||
"ph_low": "mdi:alert",
|
||||
"ph_over": "mdi:alert",
|
||||
"ph_under": "mdi:alert",
|
||||
"pump_stopped": "mdi:alert",
|
||||
"tank_level": "mdi:alert"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
{
|
||||
"domain": "neopool",
|
||||
"name": "NeoPool",
|
||||
"codeowners": ["@svasek"],
|
||||
"config_flow": true,
|
||||
"documentation": "https://www.home-assistant.io/integrations/neopool",
|
||||
"integration_type": "hub",
|
||||
"iot_class": "local_polling",
|
||||
"loggers": ["neopool_modbus"],
|
||||
"quality_scale": "silver",
|
||||
"requirements": ["neopool-modbus==3.6.0"]
|
||||
}
|
||||
@@ -0,0 +1,96 @@
|
||||
rules:
|
||||
# Bronze
|
||||
action-setup:
|
||||
status: exempt
|
||||
comment: The integration does not register any service actions.
|
||||
appropriate-polling: done
|
||||
brands: done
|
||||
common-modules: done
|
||||
config-flow-test-coverage: done
|
||||
config-flow: done
|
||||
dependency-transparency: done
|
||||
docs-actions:
|
||||
status: exempt
|
||||
comment: The integration does not register any service actions.
|
||||
docs-conditions:
|
||||
status: exempt
|
||||
comment: The integration does not provide any conditions.
|
||||
docs-high-level-description: done
|
||||
docs-installation-instructions: done
|
||||
docs-removal-instructions: done
|
||||
docs-triggers:
|
||||
status: exempt
|
||||
comment: The integration does not provide any triggers.
|
||||
entity-event-setup:
|
||||
status: exempt
|
||||
comment: |
|
||||
Entities use the coordinator pattern and do not subscribe to
|
||||
integration-specific events.
|
||||
entity-unique-id: done
|
||||
has-entity-name: done
|
||||
runtime-data: done
|
||||
test-before-configure: done
|
||||
test-before-setup: done
|
||||
unique-config-entry: done
|
||||
|
||||
# Silver
|
||||
action-exceptions:
|
||||
status: exempt
|
||||
comment: The integration does not register any service actions.
|
||||
config-entry-unloading: done
|
||||
docs-configuration-parameters: done
|
||||
docs-installation-parameters: done
|
||||
entity-unavailable: done
|
||||
integration-owner: done
|
||||
log-when-unavailable: done
|
||||
parallel-updates: done
|
||||
reauthentication-flow:
|
||||
status: exempt
|
||||
comment: Modbus TCP has no authentication mechanism.
|
||||
test-coverage: done
|
||||
|
||||
# Gold
|
||||
devices: done
|
||||
diagnostics: todo
|
||||
discovery:
|
||||
status: exempt
|
||||
comment: |
|
||||
Modbus TCP gateways have no standard discovery protocol
|
||||
(no zeroconf, SSDP, or DHCP signal that uniquely identifies
|
||||
a NeoPool controller behind the gateway).
|
||||
discovery-update-info:
|
||||
status: exempt
|
||||
comment: See discovery exemption above.
|
||||
docs-data-update: done
|
||||
docs-examples: done
|
||||
docs-known-limitations: done
|
||||
docs-supported-devices: done
|
||||
docs-supported-functions: done
|
||||
docs-troubleshooting: done
|
||||
docs-use-cases: done
|
||||
dynamic-devices:
|
||||
status: exempt
|
||||
comment: |
|
||||
One config entry maps to one physical NeoPool controller; multiple
|
||||
controllers are supported via separate config entries. The single
|
||||
device per entry is created during initial setup and cannot change.
|
||||
entity-category: done
|
||||
entity-device-class: done
|
||||
entity-disabled-by-default: done
|
||||
entity-translations: done
|
||||
exception-translations: done
|
||||
icon-translations: done
|
||||
reconfiguration-flow: todo
|
||||
repair-issues: done
|
||||
stale-devices:
|
||||
status: exempt
|
||||
comment: |
|
||||
One config entry maps to one physical device; the device is not
|
||||
removed during runtime, so there are no stale devices to clean up.
|
||||
|
||||
# Platinum
|
||||
async-dependency: done
|
||||
inject-websession:
|
||||
status: exempt
|
||||
comment: Integration uses Modbus TCP, not HTTP, so no aiohttp session is involved.
|
||||
strict-typing: done
|
||||
@@ -0,0 +1,380 @@
|
||||
"""Sensor platform for the NeoPool integration."""
|
||||
|
||||
from collections.abc import Callable
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime
|
||||
from typing import Any, override
|
||||
|
||||
from neopool_modbus.capabilities import (
|
||||
has_filtvalve,
|
||||
has_heating_relay,
|
||||
has_variable_speed_pump,
|
||||
is_chlorine_module_present,
|
||||
is_conductivity_module_present,
|
||||
is_hydrolysis_present,
|
||||
is_ionization_present,
|
||||
is_ph_module_present,
|
||||
is_redox_module_present,
|
||||
is_temperature_active,
|
||||
)
|
||||
from neopool_modbus.decoders import (
|
||||
FILTRATION_MODE_LABELS,
|
||||
FILTRATION_SPEED_STATE_LABELS,
|
||||
HIDRO_POLARITY_LABELS,
|
||||
ION_POLARITY_LABELS,
|
||||
PH_STATUS_ALARM_LABELS,
|
||||
calculate_next_interval_time,
|
||||
decode_hidro_polarity,
|
||||
decode_ion_polarity,
|
||||
decode_ph_alarm,
|
||||
decode_ph_pump_status,
|
||||
is_hydrolysis_in_percent,
|
||||
ph_pump_options,
|
||||
)
|
||||
|
||||
from homeassistant.components.sensor import (
|
||||
SensorDeviceClass,
|
||||
SensorEntity,
|
||||
SensorEntityDescription,
|
||||
SensorStateClass,
|
||||
)
|
||||
from homeassistant.const import (
|
||||
EntityCategory,
|
||||
UnitOfElectricPotential,
|
||||
UnitOfRatio,
|
||||
UnitOfTemperature,
|
||||
UnitOfTime,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
||||
|
||||
from . import NeoPoolConfigEntry
|
||||
from .coordinator import NeoPoolCoordinator
|
||||
from .entity import NeoPoolEntity
|
||||
|
||||
PARALLEL_UPDATES = 0
|
||||
|
||||
|
||||
@dataclass(frozen=True, kw_only=True)
|
||||
class NeoPoolSensorEntityDescription(SensorEntityDescription):
|
||||
"""Describes a NeoPool sensor entity."""
|
||||
|
||||
supported_fn: Callable[[dict[str, Any]], bool] | None = None
|
||||
value_fn: Callable[[dict[str, Any]], Any] | None = None
|
||||
options_fn: Callable[[dict[str, Any]], list[str]] | None = None
|
||||
unit_fn: Callable[[dict[str, Any]], str | None] | None = None
|
||||
precision_fn: Callable[[dict[str, Any]], int | None] | None = None
|
||||
|
||||
|
||||
SENSOR_DESCRIPTIONS: dict[str, NeoPoolSensorEntityDescription] = {
|
||||
"MBF_ION_CURRENT": NeoPoolSensorEntityDescription(
|
||||
key="MBF_ION_CURRENT",
|
||||
translation_key="ion_current",
|
||||
native_unit_of_measurement=UnitOfRatio.PERCENTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
supported_fn=is_ionization_present,
|
||||
),
|
||||
"MBF_HIDRO_CURRENT": NeoPoolSensorEntityDescription(
|
||||
key="MBF_HIDRO_CURRENT",
|
||||
translation_key="hidro_current",
|
||||
native_unit_of_measurement=UnitOfRatio.PERCENTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=0,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
unit_fn=lambda data: (
|
||||
UnitOfRatio.PERCENTAGE if is_hydrolysis_in_percent(data) else "g/h"
|
||||
),
|
||||
precision_fn=lambda data: 0 if is_hydrolysis_in_percent(data) else 1,
|
||||
),
|
||||
"MBF_MEASURE_PH": NeoPoolSensorEntityDescription(
|
||||
key="MBF_MEASURE_PH",
|
||||
device_class=SensorDeviceClass.PH,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
supported_fn=is_ph_module_present,
|
||||
),
|
||||
"MBF_MEASURE_RX": NeoPoolSensorEntityDescription(
|
||||
key="MBF_MEASURE_RX",
|
||||
translation_key="measure_rx",
|
||||
native_unit_of_measurement=UnitOfElectricPotential.MILLIVOLT,
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
supported_fn=is_redox_module_present,
|
||||
),
|
||||
"MBF_MEASURE_CL": NeoPoolSensorEntityDescription(
|
||||
key="MBF_MEASURE_CL",
|
||||
translation_key="measure_cl",
|
||||
native_unit_of_measurement=UnitOfRatio.PARTS_PER_MILLION,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
supported_fn=is_chlorine_module_present,
|
||||
),
|
||||
"MBF_MEASURE_CONDUCTIVITY": NeoPoolSensorEntityDescription(
|
||||
key="MBF_MEASURE_CONDUCTIVITY",
|
||||
translation_key="measure_conductivity",
|
||||
native_unit_of_measurement=UnitOfRatio.PERCENTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=0,
|
||||
supported_fn=is_conductivity_module_present,
|
||||
),
|
||||
"MBF_MEASURE_TEMPERATURE": NeoPoolSensorEntityDescription(
|
||||
key="MBF_MEASURE_TEMPERATURE",
|
||||
translation_key="measure_temperature",
|
||||
native_unit_of_measurement=UnitOfTemperature.CELSIUS,
|
||||
device_class=SensorDeviceClass.TEMPERATURE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
supported_fn=is_temperature_active,
|
||||
),
|
||||
"MBF_HIDRO_VOLTAGE": NeoPoolSensorEntityDescription(
|
||||
key="MBF_HIDRO_VOLTAGE",
|
||||
translation_key="hidro_voltage",
|
||||
native_unit_of_measurement=UnitOfElectricPotential.VOLT,
|
||||
device_class=SensorDeviceClass.VOLTAGE,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
suggested_display_precision=1,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"MBF_PAR_FILT_MODE": NeoPoolSensorEntityDescription(
|
||||
key="MBF_PAR_FILT_MODE",
|
||||
translation_key="filt_mode",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
options=list(FILTRATION_MODE_LABELS.values()),
|
||||
value_fn=lambda data: data.get("filtration_mode"),
|
||||
),
|
||||
"MBF_PH_STATUS_ALARM": NeoPoolSensorEntityDescription(
|
||||
key="MBF_PH_STATUS_ALARM",
|
||||
translation_key="ph_status_alarm",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
options=list(PH_STATUS_ALARM_LABELS.values()),
|
||||
value_fn=decode_ph_alarm,
|
||||
supported_fn=is_ph_module_present,
|
||||
),
|
||||
"HIDRO_POLARITY": NeoPoolSensorEntityDescription(
|
||||
key="HIDRO_POLARITY",
|
||||
translation_key="hidro_polarity",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
options=list(HIDRO_POLARITY_LABELS),
|
||||
value_fn=decode_hidro_polarity,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"ION_POLARITY": NeoPoolSensorEntityDescription(
|
||||
key="ION_POLARITY",
|
||||
translation_key="ion_polarity",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
options=list(ION_POLARITY_LABELS),
|
||||
value_fn=decode_ion_polarity,
|
||||
supported_fn=is_ionization_present,
|
||||
),
|
||||
"PH_PUMP_STATUS": NeoPoolSensorEntityDescription(
|
||||
key="PH_PUMP_STATUS",
|
||||
translation_key="ph_pump_status",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
options_fn=ph_pump_options,
|
||||
value_fn=decode_ph_pump_status,
|
||||
supported_fn=is_ph_module_present,
|
||||
),
|
||||
"FILTRATION_SPEED": NeoPoolSensorEntityDescription(
|
||||
key="FILTRATION_SPEED",
|
||||
translation_key="filtration_speed",
|
||||
device_class=SensorDeviceClass.ENUM,
|
||||
options=list(FILTRATION_SPEED_STATE_LABELS),
|
||||
value_fn=lambda data: data.get("filtration_speed_state"),
|
||||
supported_fn=has_variable_speed_pump,
|
||||
),
|
||||
"MBF_PAR_INTELLIGENT_INTERVALS": NeoPoolSensorEntityDescription(
|
||||
key="MBF_PAR_INTELLIGENT_INTERVALS",
|
||||
translation_key="intelligent_intervals",
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
supported_fn=lambda data: (
|
||||
has_heating_relay(data) and is_temperature_active(data)
|
||||
),
|
||||
),
|
||||
"MBF_PAR_INTELLIGENT_TT_NEXT_INTERVAL": NeoPoolSensorEntityDescription(
|
||||
key="MBF_PAR_INTELLIGENT_TT_NEXT_INTERVAL",
|
||||
translation_key="intelligent_tt_next_interval",
|
||||
device_class=SensorDeviceClass.TIMESTAMP,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
value_fn=lambda data: calculate_next_interval_time(
|
||||
data.get("MBF_PAR_INTELLIGENT_TT_NEXT_INTERVAL")
|
||||
),
|
||||
supported_fn=lambda data: (
|
||||
has_heating_relay(data) and is_temperature_active(data)
|
||||
),
|
||||
),
|
||||
"MBF_PAR_FILTVALVE_REMAINING": NeoPoolSensorEntityDescription(
|
||||
key="MBF_PAR_FILTVALVE_REMAINING",
|
||||
translation_key="filtvalve_remaining",
|
||||
native_unit_of_measurement=UnitOfTime.SECONDS,
|
||||
device_class=SensorDeviceClass.DURATION,
|
||||
state_class=SensorStateClass.MEASUREMENT,
|
||||
suggested_display_precision=0,
|
||||
supported_fn=has_filtvalve,
|
||||
),
|
||||
"CELL_RUNTIME_TOTAL": NeoPoolSensorEntityDescription(
|
||||
key="CELL_RUNTIME_TOTAL",
|
||||
translation_key="cell_runtime_total",
|
||||
native_unit_of_measurement=UnitOfTime.SECONDS,
|
||||
suggested_unit_of_measurement=UnitOfTime.HOURS,
|
||||
device_class=SensorDeviceClass.DURATION,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=0,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"CELL_RUNTIME_PART": NeoPoolSensorEntityDescription(
|
||||
key="CELL_RUNTIME_PART",
|
||||
translation_key="cell_runtime_part",
|
||||
native_unit_of_measurement=UnitOfTime.SECONDS,
|
||||
suggested_unit_of_measurement=UnitOfTime.HOURS,
|
||||
device_class=SensorDeviceClass.DURATION,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=0,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"CELL_RUNTIME_POLA": NeoPoolSensorEntityDescription(
|
||||
key="CELL_RUNTIME_POLA",
|
||||
translation_key="cell_runtime_pola",
|
||||
native_unit_of_measurement=UnitOfTime.SECONDS,
|
||||
suggested_unit_of_measurement=UnitOfTime.HOURS,
|
||||
device_class=SensorDeviceClass.DURATION,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=0,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"CELL_RUNTIME_POLB": NeoPoolSensorEntityDescription(
|
||||
key="CELL_RUNTIME_POLB",
|
||||
translation_key="cell_runtime_polb",
|
||||
native_unit_of_measurement=UnitOfTime.SECONDS,
|
||||
suggested_unit_of_measurement=UnitOfTime.HOURS,
|
||||
device_class=SensorDeviceClass.DURATION,
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=0,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
"CELL_RUNTIME_POL_CHANGES": NeoPoolSensorEntityDescription(
|
||||
key="CELL_RUNTIME_POL_CHANGES",
|
||||
translation_key="cell_runtime_pol_changes",
|
||||
state_class=SensorStateClass.TOTAL_INCREASING,
|
||||
suggested_display_precision=0,
|
||||
entity_category=EntityCategory.DIAGNOSTIC,
|
||||
entity_registry_enabled_default=False,
|
||||
supported_fn=is_hydrolysis_present,
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
async def async_setup_entry(
|
||||
hass: HomeAssistant,
|
||||
entry: NeoPoolConfigEntry,
|
||||
async_add_entities: AddConfigEntryEntitiesCallback,
|
||||
) -> None:
|
||||
"""Set up NeoPool sensors from a config entry."""
|
||||
coordinator = entry.runtime_data
|
||||
|
||||
async_add_entities(
|
||||
NeoPoolSensor(coordinator, key, desc)
|
||||
for key, desc in SENSOR_DESCRIPTIONS.items()
|
||||
if desc.supported_fn is None or desc.supported_fn(coordinator.data)
|
||||
)
|
||||
|
||||
|
||||
_PRODUCTION_KEYS_REQUIRING_FILTRATION = frozenset(
|
||||
{
|
||||
"MBF_HIDRO_CURRENT",
|
||||
"MBF_HIDRO_VOLTAGE",
|
||||
"MBF_ION_CURRENT",
|
||||
}
|
||||
)
|
||||
|
||||
_MEASURE_KEYS_REQUIRING_FILTRATION = frozenset(
|
||||
{
|
||||
"MBF_MEASURE_TEMPERATURE",
|
||||
"MBF_MEASURE_PH",
|
||||
"MBF_MEASURE_RX",
|
||||
"MBF_MEASURE_CL",
|
||||
"MBF_MEASURE_CONDUCTIVITY",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
class NeoPoolSensor(NeoPoolEntity, SensorEntity):
|
||||
"""Representation of a NeoPool sensor."""
|
||||
|
||||
entity_description: NeoPoolSensorEntityDescription
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
coordinator: NeoPoolCoordinator,
|
||||
key: str,
|
||||
description: NeoPoolSensorEntityDescription,
|
||||
) -> None:
|
||||
"""Initialize the NeoPool sensor entity."""
|
||||
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 suggested_display_precision(self) -> int | None:
|
||||
"""Return the suggested display precision for the sensor value."""
|
||||
if (precision_fn := self.entity_description.precision_fn) is not None:
|
||||
return precision_fn(self.coordinator.data)
|
||||
return super().suggested_display_precision
|
||||
|
||||
@property
|
||||
@override
|
||||
def native_unit_of_measurement(self) -> str | None:
|
||||
"""Return the unit of measurement for the sensor value."""
|
||||
if (unit_fn := self.entity_description.unit_fn) is not None:
|
||||
return unit_fn(self.coordinator.data)
|
||||
return super().native_unit_of_measurement
|
||||
|
||||
def _filtration_off(self) -> bool:
|
||||
"""Return True when the filtration pump is off."""
|
||||
return self.coordinator.data.get("Filtration Pump") is False
|
||||
|
||||
def _is_measurement_suppressed(self) -> bool:
|
||||
"""Return True if a measurement sensor should report None."""
|
||||
if self._key not in _MEASURE_KEYS_REQUIRING_FILTRATION:
|
||||
return False
|
||||
return self._filtration_off()
|
||||
|
||||
def _is_production_suppressed(self) -> bool:
|
||||
"""Return True if a production sensor should report 0."""
|
||||
if self._key not in _PRODUCTION_KEYS_REQUIRING_FILTRATION:
|
||||
return False
|
||||
return self._filtration_off()
|
||||
|
||||
@property
|
||||
@override
|
||||
def native_value(self) -> float | int | str | datetime | None:
|
||||
"""Return the actual sensor value from coordinator data."""
|
||||
if self._is_measurement_suppressed():
|
||||
return None
|
||||
if self._is_production_suppressed():
|
||||
return 0
|
||||
if (value_fn := self.entity_description.value_fn) is not None:
|
||||
value: float | int | str | datetime | None = value_fn(self.coordinator.data)
|
||||
return value
|
||||
return self.coordinator.data.get(self._key)
|
||||
|
||||
@property
|
||||
@override
|
||||
def options(self) -> list[str] | None:
|
||||
"""Return the list of options for the sensor."""
|
||||
if (options_fn := self.entity_description.options_fn) is not None:
|
||||
return options_fn(self.coordinator.data)
|
||||
return super().options
|
||||
@@ -0,0 +1,150 @@
|
||||
{
|
||||
"config": {
|
||||
"abort": {
|
||||
"already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
|
||||
},
|
||||
"error": {
|
||||
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
|
||||
"cannot_read_modbus": "Connected, but cannot read from the Modbus device. Check unit ID and framer settings."
|
||||
},
|
||||
"step": {
|
||||
"user": {
|
||||
"data": {
|
||||
"host": "[%key:common::config_flow::data::host%]",
|
||||
"modbus_framer": "Modbus framer",
|
||||
"port": "[%key:common::config_flow::data::port%]",
|
||||
"unit_id": "Unit ID"
|
||||
},
|
||||
"data_description": {
|
||||
"host": "Enter the IP address of the Modbus TCP gateway connected to your pool controller.",
|
||||
"modbus_framer": "Wire protocol used by the gateway. Select TCP for MBAP framing (default for Ethernet-native gateways). Select RTU for RTU framing tunnelled through a TCP socket, used by passthrough serial-to-TCP bridges like ESPHome's stream_server.",
|
||||
"port": "Standard Modbus TCP port (default: 502). Change only if your gateway uses a non-standard port.",
|
||||
"unit_id": "Modbus device address (1-247). Identifies the specific device on a shared bus. Most direct-connected devices use 1. Gateways may route requests to different addresses."
|
||||
},
|
||||
"description": "Configure the connection to your NeoPool controller.",
|
||||
"title": "NeoPool Connection"
|
||||
}
|
||||
}
|
||||
},
|
||||
"entity": {
|
||||
"sensor": {
|
||||
"cell_runtime_part": {
|
||||
"name": "Cell runtime since reset"
|
||||
},
|
||||
"cell_runtime_pol_changes": {
|
||||
"name": "Cell polarity changes"
|
||||
},
|
||||
"cell_runtime_pola": {
|
||||
"name": "Cell runtime in polarity 1"
|
||||
},
|
||||
"cell_runtime_polb": {
|
||||
"name": "Cell runtime in polarity 2"
|
||||
},
|
||||
"cell_runtime_total": {
|
||||
"name": "Cell runtime total"
|
||||
},
|
||||
"filt_mode": {
|
||||
"name": "Filtration mode",
|
||||
"state": {
|
||||
"auto": "Automatic",
|
||||
"backwash": "Backwash",
|
||||
"heating": "Heating",
|
||||
"intelligent": "Intelligent",
|
||||
"manual": "[%key:common::state::manual%]",
|
||||
"smart": "Smart"
|
||||
}
|
||||
},
|
||||
"filtration_speed": {
|
||||
"name": "Current filtration speed",
|
||||
"state": {
|
||||
"high": "[%key:common::state::high%]",
|
||||
"low": "[%key:common::state::low%]",
|
||||
"mid": "Medium",
|
||||
"off": "[%key:common::state::off%]"
|
||||
}
|
||||
},
|
||||
"filtvalve_remaining": {
|
||||
"name": "Backwash time remaining"
|
||||
},
|
||||
"hidro_current": {
|
||||
"name": "Hydrolysis intensity"
|
||||
},
|
||||
"hidro_polarity": {
|
||||
"name": "Hydrolysis polarity",
|
||||
"state": {
|
||||
"dead_time": "Dead time",
|
||||
"no_flow": "No flow",
|
||||
"off": "[%key:common::state::off%]",
|
||||
"pol1": "Polarity 1",
|
||||
"pol2": "Polarity 2"
|
||||
}
|
||||
},
|
||||
"hidro_voltage": {
|
||||
"name": "Hydrolysis voltage"
|
||||
},
|
||||
"intelligent_intervals": {
|
||||
"name": "Intelligent mode intervals"
|
||||
},
|
||||
"intelligent_tt_next_interval": {
|
||||
"name": "Intelligent mode next interval start"
|
||||
},
|
||||
"ion_current": {
|
||||
"name": "Ionization level"
|
||||
},
|
||||
"ion_polarity": {
|
||||
"name": "Ionizer polarity",
|
||||
"state": {
|
||||
"dead_time": "Dead time",
|
||||
"off": "[%key:common::state::off%]",
|
||||
"pol1": "Polarity 1",
|
||||
"pol2": "Polarity 2"
|
||||
}
|
||||
},
|
||||
"measure_cl": {
|
||||
"name": "Salt level"
|
||||
},
|
||||
"measure_conductivity": {
|
||||
"name": "Conductivity level"
|
||||
},
|
||||
"measure_rx": {
|
||||
"name": "Redox potential"
|
||||
},
|
||||
"measure_temperature": {
|
||||
"name": "Water temperature"
|
||||
},
|
||||
"ph_pump_status": {
|
||||
"name": "pH pump status",
|
||||
"state": {
|
||||
"acid": "Acid pump",
|
||||
"base": "Base pump",
|
||||
"both": "Both pumps",
|
||||
"idle": "[%key:common::state::idle%]",
|
||||
"off": "[%key:common::state::off%]"
|
||||
}
|
||||
},
|
||||
"ph_status_alarm": {
|
||||
"name": "pH alarm",
|
||||
"state": {
|
||||
"ok": "OK",
|
||||
"ph_high": "pH too high",
|
||||
"ph_low": "pH too low",
|
||||
"ph_over": "pH higher than the set point",
|
||||
"ph_under": "pH lower than the set point",
|
||||
"pump_stopped": "Pump stopped (exceeded working time)",
|
||||
"tank_level": "Tank level alarm"
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
"exceptions": {
|
||||
"modbus_communication_error": {
|
||||
"message": "An error occurred while communicating with the NeoPool controller: {error}"
|
||||
}
|
||||
},
|
||||
"issues": {
|
||||
"corrupted_gpio": {
|
||||
"description": "The following GPIO register(s) on your pool controller contain invalid values:\n\n{details}\n\nThis typically happens when the Modbus gateway framing mode does not match the integration's framer setting. The affected function(s) will not work correctly until the register(s) are restored to valid values.\n\nSee the integration documentation for repair instructions.",
|
||||
"title": "Corrupted GPIO register(s) detected"
|
||||
}
|
||||
}
|
||||
}
|
||||
Generated
+1
@@ -498,6 +498,7 @@ FLOWS = {
|
||||
"nasweb",
|
||||
"neato",
|
||||
"nederlandse_spoorwegen",
|
||||
"neopool",
|
||||
"ness_alarm",
|
||||
"nest",
|
||||
"netatmo",
|
||||
|
||||
@@ -4630,6 +4630,12 @@
|
||||
"integration_type": "virtual",
|
||||
"supported_by": "shelly"
|
||||
},
|
||||
"neopool": {
|
||||
"name": "NeoPool",
|
||||
"integration_type": "hub",
|
||||
"config_flow": true,
|
||||
"iot_class": "local_polling"
|
||||
},
|
||||
"ness_alarm": {
|
||||
"name": "Ness Alarm",
|
||||
"integration_type": "hub",
|
||||
|
||||
@@ -3707,6 +3707,16 @@ disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.neopool.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
disallow_subclassing_any = true
|
||||
disallow_untyped_calls = true
|
||||
disallow_untyped_decorators = true
|
||||
disallow_untyped_defs = true
|
||||
warn_return_any = true
|
||||
warn_unreachable = true
|
||||
|
||||
[mypy-homeassistant.components.nest.*]
|
||||
check_untyped_defs = true
|
||||
disallow_incomplete_defs = true
|
||||
|
||||
Generated
+3
@@ -1639,6 +1639,9 @@ nad-receiver==0.3.0
|
||||
# homeassistant.components.keenetic_ndms2
|
||||
ndms2-client==0.1.2
|
||||
|
||||
# homeassistant.components.neopool
|
||||
neopool-modbus==3.6.0
|
||||
|
||||
# homeassistant.components.ness_alarm
|
||||
nessclient==1.3.1
|
||||
|
||||
|
||||
@@ -0,0 +1,14 @@
|
||||
"""Tests for the NeoPool integration."""
|
||||
|
||||
from homeassistant.core import HomeAssistant
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
|
||||
async def setup_integration(
|
||||
hass: HomeAssistant, mock_config_entry: MockConfigEntry
|
||||
) -> None:
|
||||
"""Set up the NeoPool integration for testing."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
await hass.config_entries.async_setup(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
@@ -0,0 +1,175 @@
|
||||
"""Common fixtures for the NeoPool tests."""
|
||||
|
||||
from collections.abc import Generator
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.neopool.const import (
|
||||
CURRENT_VERSION,
|
||||
DEFAULT_PORT,
|
||||
DEFAULT_UNIT_ID,
|
||||
DOMAIN,
|
||||
)
|
||||
from homeassistant.const import CONF_HOST, CONF_NAME, CONF_PORT
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
MOCK_HOST = "192.0.2.1"
|
||||
MOCK_PORT = DEFAULT_PORT
|
||||
MOCK_NAME = "Pool"
|
||||
MOCK_SERIAL = "1234567890"
|
||||
|
||||
|
||||
MOCK_POOL_DATA: dict[str, Any] = {
|
||||
"MBF_POWER_MODULE_VERSION": 0x1234,
|
||||
"MBF_PAR_VERSION": 0x100,
|
||||
"MBF_PAR_MODEL": 0x0003,
|
||||
"MBF_PAR_SERNUM": int(MOCK_SERIAL),
|
||||
"MBF_PAR_FILTRATION_CONF": 1,
|
||||
"MBF_PAR_FILT_GPIO": 1,
|
||||
"MBF_PAR_LIGHTING_GPIO": 2,
|
||||
"MBF_PAR_HEATING_GPIO": 3,
|
||||
"MBF_PAR_PH_ACID_RELAY_GPIO": 4,
|
||||
"MBF_PAR_PH_BASE_RELAY_GPIO": 5,
|
||||
"MBF_PAR_RX_RELAY_GPIO": 6,
|
||||
"MBF_PAR_CL_RELAY_GPIO": 7,
|
||||
"MBF_PAR_CD_RELAY_GPIO": 0,
|
||||
"MBF_PAR_UV_RELAY_GPIO": 1,
|
||||
"MBF_PAR_FILTVALVE_GPIO": 1,
|
||||
"MBF_PAR_FILTVALVE_ENABLE": 1,
|
||||
"MBF_PAR_TEMPERATURE_ACTIVE": 1,
|
||||
"MBF_PAR_UICFG_MACHINE": 0,
|
||||
"MBF_PAR_RELAY_PH": 0,
|
||||
"Hydrolysis module detected": True,
|
||||
"Redox measurement module detected": True,
|
||||
"pH measurement module detected": True,
|
||||
"Chlorine measurement module detected": True,
|
||||
"Conductivity measurement module detected": True,
|
||||
"Ionization module detected": True,
|
||||
"MBF_PAR_FILT_MODE": 0,
|
||||
"filtration_mode": "manual",
|
||||
"filtration_speed_state": "off",
|
||||
"MBF_MEASURE_TEMPERATURE": 250,
|
||||
"MBF_MEASURE_PH": 720,
|
||||
"MBF_MEASURE_RX": 650,
|
||||
"MBF_MEASURE_CL": 120,
|
||||
"MBF_MEASURE_CONDUCTIVITY": 45,
|
||||
"MBF_HIDRO_CURRENT": 70,
|
||||
"MBF_HIDRO_VOLTAGE": 24,
|
||||
"MBF_ION_CURRENT": 50,
|
||||
"MBF_PAR_INTELLIGENT_INTERVALS": 4,
|
||||
"MBF_PAR_INTELLIGENT_TT_NEXT_INTERVAL": 7200,
|
||||
"MBF_PAR_FILTVALVE_REMAINING": 0,
|
||||
"HIDRO_POLARITY": 0,
|
||||
"ION_POLARITY": 0,
|
||||
"PH_PUMP_STATUS": "off",
|
||||
"HIDRO in Pol1": False,
|
||||
"HIDRO in Pol2": False,
|
||||
"HIDRO in dead time": False,
|
||||
"ION in Pol1": False,
|
||||
"ION in Pol2": False,
|
||||
"ION in dead time": False,
|
||||
"pH control module": True,
|
||||
"pH pump active": False,
|
||||
"pH acid pump active": False,
|
||||
"Filtration Pump": False,
|
||||
"MBF_PAR_HIDRO_COVER_REDUCTION": 0x0C19,
|
||||
"Pool Cover": 0,
|
||||
"CELL_RUNTIME_TOTAL": 0x00010000,
|
||||
"CELL_RUNTIME_PART": 0x00000E10,
|
||||
"CELL_RUNTIME_POLA": 0x00000708,
|
||||
"CELL_RUNTIME_POLB": 0x00000708,
|
||||
"CELL_RUNTIME_POL_CHANGES": 0x00000007,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_setup_entry() -> Generator[AsyncMock]:
|
||||
"""Override async_setup_entry."""
|
||||
with patch(
|
||||
"homeassistant.components.neopool.async_setup_entry", return_value=True
|
||||
) as mock_setup_entry:
|
||||
yield mock_setup_entry
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_entry() -> MockConfigEntry:
|
||||
"""Return a config entry with a bare-serial unique_id."""
|
||||
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",
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_neopool_client() -> Generator[MagicMock]:
|
||||
"""Patch the NeoPoolModbusClient and return a configurable mock instance."""
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.components.neopool.NeoPoolModbusClient",
|
||||
autospec=True,
|
||||
) as mock_client_cls,
|
||||
patch(
|
||||
"homeassistant.components.neopool.config_flow.async_probe_serial",
|
||||
new=AsyncMock(return_value=MOCK_SERIAL),
|
||||
),
|
||||
):
|
||||
mock_client = mock_client_cls.return_value
|
||||
mock_client.async_read_all = AsyncMock(return_value=dict(MOCK_POOL_DATA))
|
||||
mock_client.close = AsyncMock()
|
||||
yield mock_client
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def minimal_pool_data() -> dict[str, Any]:
|
||||
"""Pool data with all optional capability flags off.
|
||||
|
||||
Used to drive the 'should-skip' branches for supported_fn gating.
|
||||
"""
|
||||
return {
|
||||
"MBF_POWER_MODULE_VERSION": 0x1234,
|
||||
"MBF_PAR_VERSION": 0x100,
|
||||
"MBF_PAR_MODEL": 0,
|
||||
"MBF_PAR_SERNUM": int(MOCK_SERIAL),
|
||||
"MBF_PAR_FILTRATION_CONF": 0,
|
||||
"MBF_PAR_FILT_GPIO": 0,
|
||||
"MBF_PAR_LIGHTING_GPIO": 0,
|
||||
"MBF_PAR_HEATING_GPIO": 0,
|
||||
"MBF_PAR_PH_ACID_RELAY_GPIO": 0,
|
||||
"MBF_PAR_PH_BASE_RELAY_GPIO": 0,
|
||||
"MBF_PAR_RX_RELAY_GPIO": 0,
|
||||
"MBF_PAR_CL_RELAY_GPIO": 0,
|
||||
"MBF_PAR_CD_RELAY_GPIO": 0,
|
||||
"MBF_PAR_UV_RELAY_GPIO": 0,
|
||||
"MBF_PAR_FILTVALVE_GPIO": 0,
|
||||
"MBF_PAR_FILTVALVE_ENABLE": 0,
|
||||
"MBF_PAR_TEMPERATURE_ACTIVE": 0,
|
||||
"Hydrolysis module detected": False,
|
||||
"Redox measurement module detected": False,
|
||||
"pH measurement module detected": False,
|
||||
"MBF_PAR_FILT_MODE": 0,
|
||||
"filtration_mode": "manual",
|
||||
"filtration_speed_state": "off",
|
||||
"Filtration Pump": False,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_socket_connection() -> Generator[AsyncMock]:
|
||||
"""Patch the lib probe in config_flow so we don't hit the network."""
|
||||
with patch(
|
||||
"homeassistant.components.neopool.config_flow.async_probe_serial",
|
||||
new=AsyncMock(return_value=MOCK_SERIAL),
|
||||
) as mock:
|
||||
yield mock
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,106 @@
|
||||
"""Test the NeoPool config flow."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
from neopool_modbus.exceptions import (
|
||||
NeoPoolConnectionError,
|
||||
NeoPoolModbusError,
|
||||
NeoPoolTimeoutError,
|
||||
)
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.neopool.const import DEFAULT_UNIT_ID, DOMAIN
|
||||
from homeassistant.config_entries import SOURCE_USER
|
||||
from homeassistant.const import CONF_HOST, CONF_PORT
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.data_entry_flow import FlowResultType
|
||||
|
||||
from .conftest import MOCK_HOST, MOCK_PORT, MOCK_SERIAL
|
||||
|
||||
from tests.common import MockConfigEntry
|
||||
|
||||
USER_INPUT = {
|
||||
CONF_HOST: MOCK_HOST,
|
||||
CONF_PORT: MOCK_PORT,
|
||||
"unit_id": DEFAULT_UNIT_ID,
|
||||
"modbus_framer": "tcp",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_neopool_client")
|
||||
async def test_user_flow(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
) -> None:
|
||||
"""Test a happy-path config flow creates the entry."""
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["step_id"] == "user"
|
||||
assert not result["errors"]
|
||||
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
assert result["title"] == MOCK_HOST
|
||||
assert result["data"][CONF_HOST] == MOCK_HOST
|
||||
assert result["data"][CONF_PORT] == MOCK_PORT
|
||||
assert result["result"].unique_id == MOCK_SERIAL
|
||||
assert mock_setup_entry.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("exc_cls", "error_key"),
|
||||
[
|
||||
(NeoPoolConnectionError, "cannot_connect"),
|
||||
(NeoPoolTimeoutError, "cannot_connect"),
|
||||
(NeoPoolModbusError, "cannot_read_modbus"),
|
||||
],
|
||||
)
|
||||
async def test_user_flow_probe_errors_recover(
|
||||
hass: HomeAssistant,
|
||||
mock_setup_entry: AsyncMock,
|
||||
mock_socket_connection: AsyncMock,
|
||||
exc_cls: type[Exception],
|
||||
error_key: str,
|
||||
) -> None:
|
||||
"""Probe errors surface as form errors, and the flow recovers on retry."""
|
||||
mock_socket_connection.side_effect = exc_cls("boom")
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.FORM
|
||||
assert result["errors"] == {CONF_HOST: error_key}
|
||||
|
||||
mock_socket_connection.side_effect = None
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], USER_INPUT
|
||||
)
|
||||
assert result["type"] is FlowResultType.CREATE_ENTRY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_neopool_client")
|
||||
async def test_user_flow_already_configured(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Test config flow aborts when the same device is already configured."""
|
||||
mock_config_entry.add_to_hass(hass)
|
||||
|
||||
result = await hass.config_entries.flow.async_init(
|
||||
DOMAIN, context={"source": SOURCE_USER}
|
||||
)
|
||||
result = await hass.config_entries.flow.async_configure(
|
||||
result["flow_id"], USER_INPUT
|
||||
)
|
||||
|
||||
assert result["type"] is FlowResultType.ABORT
|
||||
assert result["reason"] == "already_configured"
|
||||
@@ -0,0 +1,192 @@
|
||||
"""Test the NeoPool integration setup, unload, and lifecycle."""
|
||||
|
||||
from datetime import timedelta
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from freezegun.api import FrozenDateTimeFactory
|
||||
from neopool_modbus.registers import MAX_RELAY_GPIO
|
||||
import pytest
|
||||
|
||||
from homeassistant.components.neopool.const import DOMAIN
|
||||
from homeassistant.config_entries import ConfigEntryState
|
||||
from homeassistant.const import STATE_UNAVAILABLE
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers import device_registry as dr, issue_registry as ir
|
||||
|
||||
from . import setup_integration
|
||||
from .conftest import MOCK_POOL_DATA, MOCK_SERIAL
|
||||
|
||||
from tests.common import MockConfigEntry, async_fire_time_changed
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_neopool_client")
|
||||
async def test_setup_and_unload(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Set up the integration end-to-end and tear it down again."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
assert await hass.config_entries.async_unload(mock_config_entry.entry_id)
|
||||
await hass.async_block_till_done()
|
||||
assert mock_config_entry.state is ConfigEntryState.NOT_LOADED
|
||||
|
||||
|
||||
async def test_setup_first_refresh_fails_marks_retry(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
) -> None:
|
||||
"""Setup re-tries when the first Modbus read raises."""
|
||||
mock_neopool_client.async_read_all = AsyncMock(
|
||||
side_effect=ConnectionError("Modbus down")
|
||||
)
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
assert mock_config_entry.state is ConfigEntryState.SETUP_RETRY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_neopool_client")
|
||||
async def test_device_registered_with_firmware(
|
||||
hass: HomeAssistant,
|
||||
device_registry: dr.DeviceRegistry,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""The first successful read populates firmware on the device entry."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
device = device_registry.async_get_device(identifiers={(DOMAIN, MOCK_SERIAL)})
|
||||
assert device is not None
|
||||
assert "18.52" in (device.sw_version or "")
|
||||
|
||||
|
||||
async def test_transient_modbus_failure_after_first_success_marks_unavailable(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Entities transition to unavailable when polling fails after a good read."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
entity_id = "sensor.neopool_water_temperature"
|
||||
assert hass.states.get(entity_id).state != STATE_UNAVAILABLE
|
||||
|
||||
mock_neopool_client.async_read_all.side_effect = ConnectionError("Modbus fail")
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
assert hass.states.get(entity_id).state == STATE_UNAVAILABLE
|
||||
assert mock_config_entry.state is ConfigEntryState.LOADED
|
||||
|
||||
|
||||
async def test_corrupt_gpio_creates_repair_issue(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
) -> None:
|
||||
"""A GPIO register outside 0..MAX_RELAY_GPIO opens a corrupted_gpio issue."""
|
||||
bad_data = dict(MOCK_POOL_DATA)
|
||||
bad_data["MBF_PAR_FILT_GPIO"] = MAX_RELAY_GPIO + 1
|
||||
mock_neopool_client.async_read_all = AsyncMock(return_value=bad_data)
|
||||
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
issue = issue_registry.async_get_issue(DOMAIN, "corrupted_gpio")
|
||||
assert issue is not None
|
||||
assert issue.severity is ir.IssueSeverity.ERROR
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_neopool_client")
|
||||
async def test_clean_gpio_does_not_create_issue(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
) -> None:
|
||||
"""A clean read does not open a corrupted_gpio issue."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
assert issue_registry.async_get_issue(DOMAIN, "corrupted_gpio") is None
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_neopool_client")
|
||||
async def test_corrupt_gpio_clears_stale_issue_from_previous_session(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
) -> None:
|
||||
"""A stale issue from a previous session clears on the first clean poll."""
|
||||
ir.async_create_issue(
|
||||
hass,
|
||||
DOMAIN,
|
||||
"corrupted_gpio",
|
||||
is_fixable=False,
|
||||
severity=ir.IssueSeverity.ERROR,
|
||||
translation_key="corrupted_gpio",
|
||||
translation_placeholders={"details": "- stale"},
|
||||
)
|
||||
assert issue_registry.async_get_issue(DOMAIN, "corrupted_gpio") is not None
|
||||
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
assert issue_registry.async_get_issue(DOMAIN, "corrupted_gpio") is None
|
||||
|
||||
|
||||
async def test_corrupt_gpio_logs_error_only_on_state_change(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
caplog: pytest.LogCaptureFixture,
|
||||
) -> None:
|
||||
"""The ERROR log fires on entering corruption and clears on healing."""
|
||||
bad_data = dict(MOCK_POOL_DATA)
|
||||
bad_data["MBF_PAR_FILT_GPIO"] = MAX_RELAY_GPIO + 1
|
||||
mock_neopool_client.async_read_all = AsyncMock(return_value=bad_data)
|
||||
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
assert sum("Corrupted GPIO register" in r.message for r in caplog.records) == 1
|
||||
|
||||
caplog.clear()
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert not any("Corrupted GPIO register" in r.message for r in caplog.records)
|
||||
|
||||
mock_neopool_client.async_read_all = AsyncMock(return_value=dict(MOCK_POOL_DATA))
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
assert issue_registry.async_get_issue(DOMAIN, "corrupted_gpio") is None
|
||||
|
||||
|
||||
async def test_corrupt_gpio_updates_issue_on_value_change(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
issue_registry: ir.IssueRegistry,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""The repair issue details refresh when a corrupted register value changes."""
|
||||
first = dict(MOCK_POOL_DATA)
|
||||
first["MBF_PAR_FILT_GPIO"] = MAX_RELAY_GPIO + 1
|
||||
mock_neopool_client.async_read_all = AsyncMock(return_value=first)
|
||||
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
issue = issue_registry.async_get_issue(DOMAIN, "corrupted_gpio")
|
||||
assert issue is not None
|
||||
assert issue.translation_placeholders is not None
|
||||
assert str(MAX_RELAY_GPIO + 1) in issue.translation_placeholders["details"]
|
||||
|
||||
second = dict(MOCK_POOL_DATA)
|
||||
second["MBF_PAR_FILT_GPIO"] = MAX_RELAY_GPIO + 2
|
||||
mock_neopool_client.async_read_all = AsyncMock(return_value=second)
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
issue = issue_registry.async_get_issue(DOMAIN, "corrupted_gpio")
|
||||
assert issue is not None
|
||||
assert issue.translation_placeholders is not None
|
||||
assert str(MAX_RELAY_GPIO + 2) in issue.translation_placeholders["details"]
|
||||
@@ -0,0 +1,242 @@
|
||||
"""Tests for the NeoPool sensor platform."""
|
||||
|
||||
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.sensor import ATTR_OPTIONS
|
||||
from homeassistant.const import ATTR_UNIT_OF_MEASUREMENT, 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
|
||||
|
||||
|
||||
async def test_measurement_sensors_suppressed_when_filtration_off(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Probe sensors report unknown while filtration pump is off (stale reading)."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
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()
|
||||
for entity_id in (
|
||||
"sensor.neopool_ph",
|
||||
"sensor.neopool_redox_potential",
|
||||
"sensor.neopool_water_temperature",
|
||||
):
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None, f"{entity_id} not registered"
|
||||
assert state.state == "unknown"
|
||||
|
||||
|
||||
async def test_production_sensors_zero_when_filtration_off(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""Production sensors report 0 while filtration pump is off (cell idle)."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
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()
|
||||
for entity_id in (
|
||||
"sensor.neopool_hydrolysis_intensity",
|
||||
"sensor.neopool_ionization_level",
|
||||
):
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None, f"{entity_id} not registered"
|
||||
assert state.state == "0"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("filt_mode", "expected"),
|
||||
[
|
||||
(0, "manual"),
|
||||
(1, "auto"),
|
||||
(2, "heating"),
|
||||
(3, "smart"),
|
||||
(4, "intelligent"),
|
||||
(13, "backwash"),
|
||||
],
|
||||
)
|
||||
async def test_filt_mode_native_value(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
filt_mode: int,
|
||||
expected: str,
|
||||
) -> None:
|
||||
"""Filt mode native value reads the lib's decoded filtration_mode key."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"MBF_PAR_FILT_MODE": filt_mode,
|
||||
"filtration_mode": expected,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get("sensor.neopool_filtration_mode")
|
||||
assert state is not None
|
||||
assert state.state == expected
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("relay", "expected_options"),
|
||||
[
|
||||
pytest.param(1, ["off", "idle", "acid"], id="acid_only"),
|
||||
pytest.param(2, ["off", "idle", "base"], id="base_only"),
|
||||
pytest.param(0, ["off", "idle", "acid", "base", "both"], id="both_relays"),
|
||||
],
|
||||
)
|
||||
async def test_ph_pump_status_options_per_relay_config(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
relay: int,
|
||||
expected_options: list[str],
|
||||
) -> None:
|
||||
"""The pH pump status options list shrinks based on the relay configuration."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"MBF_PAR_RELAY_PH": relay,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
state = hass.states.get("sensor.neopool_ph_pump_status")
|
||||
assert state is not None
|
||||
assert state.attributes[ATTR_OPTIONS] == expected_options
|
||||
|
||||
|
||||
async def test_hidro_current_g_per_hour_mode(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
freezer: FrozenDateTimeFactory,
|
||||
) -> None:
|
||||
"""In g/h mode HIDRO_CURRENT swaps unit and bumps display precision."""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
mock_neopool_client.async_read_all.return_value = {
|
||||
**MOCK_POOL_DATA,
|
||||
"MBF_PAR_UICFG_MACHINE": 1,
|
||||
}
|
||||
freezer.tick(timedelta(seconds=60))
|
||||
async_fire_time_changed(hass)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
state = hass.states.get("sensor.neopool_hydrolysis_intensity")
|
||||
assert state is not None
|
||||
assert state.attributes[ATTR_UNIT_OF_MEASUREMENT] == "g/h"
|
||||
|
||||
|
||||
_CELL_RUNTIME_ENTITY_IDS: dict[str, str] = {
|
||||
"CELL_RUNTIME_TOTAL": "sensor.neopool_cell_runtime_total",
|
||||
"CELL_RUNTIME_PART": "sensor.neopool_cell_runtime_since_reset",
|
||||
"CELL_RUNTIME_POLA": "sensor.neopool_cell_runtime_in_polarity_1",
|
||||
"CELL_RUNTIME_POLB": "sensor.neopool_cell_runtime_in_polarity_2",
|
||||
"CELL_RUNTIME_POL_CHANGES": "sensor.neopool_cell_polarity_changes",
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("key", "expected_seconds"),
|
||||
[
|
||||
("CELL_RUNTIME_TOTAL", 65536),
|
||||
("CELL_RUNTIME_PART", 3600),
|
||||
("CELL_RUNTIME_POLA", 1800),
|
||||
("CELL_RUNTIME_POLB", 1800),
|
||||
],
|
||||
)
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_neopool_client")
|
||||
async def test_cell_runtime_duration_sensor_reads_combined_register(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
key: str,
|
||||
expected_seconds: int,
|
||||
) -> None:
|
||||
"""Each duration CELL_RUNTIME_* sensor reads the combined u32 key from coordinator data.
|
||||
|
||||
Sensors have ``entity_registry_enabled_default=False``; enabling every
|
||||
disabled-by-default entity via the fixture avoids the reload dance. The
|
||||
sensors declare seconds but suggest hours, so ``state.state`` is expressed
|
||||
in hours (converted by the frontend layer).
|
||||
"""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
entity_id = _CELL_RUNTIME_ENTITY_IDS[key]
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None, f"{entity_id} not registered"
|
||||
assert float(state.state) == pytest.approx(expected_seconds / 3600, abs=1e-4)
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("entity_registry_enabled_by_default", "mock_neopool_client")
|
||||
async def test_cell_runtime_pol_changes_sensor_reads_combined_register(
|
||||
hass: HomeAssistant,
|
||||
mock_config_entry: MockConfigEntry,
|
||||
) -> None:
|
||||
"""CELL_RUNTIME_POL_CHANGES reads the combined u32 key as a raw counter.
|
||||
|
||||
Unlike the duration sensors, this one has no unit and no unit conversion,
|
||||
so ``state.state`` is the raw integer from coordinator data.
|
||||
"""
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
|
||||
entity_id = _CELL_RUNTIME_ENTITY_IDS["CELL_RUNTIME_POL_CHANGES"]
|
||||
state = hass.states.get(entity_id)
|
||||
assert state is not None, f"{entity_id} not registered"
|
||||
assert state.state == "7"
|
||||
|
||||
|
||||
@pytest.mark.freeze_time("2026-07-03T12:00:00Z")
|
||||
@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: MockConfigEntry,
|
||||
) -> None:
|
||||
"""Snapshot every entity registered by the sensor platform."""
|
||||
with patch("homeassistant.components.neopool.PLATFORMS", [Platform.SENSOR]):
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
|
||||
|
||||
@pytest.mark.freeze_time("2026-07-03T12:00:00Z")
|
||||
@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: MockConfigEntry,
|
||||
mock_neopool_client: MagicMock,
|
||||
minimal_pool_data: dict[str, Any],
|
||||
) -> None:
|
||||
"""Snapshot the 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.SENSOR]):
|
||||
await setup_integration(hass, mock_config_entry)
|
||||
await snapshot_platform(hass, entity_registry, snapshot, mock_config_entry.entry_id)
|
||||
Reference in New Issue
Block a user