Add time platform to NeoPool (#182432)

homeassistant/components/neopool/__init__.py

@@ -28,6 +28,11 @@ async def async_setup_entry(hass: HomeAssistant, entry: NeoPoolConfigEntry) -> b
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
 
+    # The first refresh ran before any entity registered its context, so
+    # context-gated timer blocks were skipped; seed one more read now that
+    # every context exists instead of waiting for the next scheduled poll.
+    await coordinator.async_refresh()
+
     return True
 
 

homeassistant/components/neopool/const.py

@@ -12,6 +12,7 @@ PLATFORMS: list[Platform] = [
     Platform.NUMBER,
     Platform.SENSOR,
     Platform.SWITCH,
+    Platform.TIME,
 ]
 
 DEFAULT_SCAN_INTERVAL = 20  # in seconds

homeassistant/components/neopool/coordinator.py

@@ -1,6 +1,7 @@
 """Data update coordinator for the NeoPool integration."""
 
 import asyncio
+from collections import defaultdict
 from datetime import timedelta
 import logging
 from typing import Any, override
@@ -9,6 +10,7 @@ from neopool_modbus import NeoPoolModbusClient
 from neopool_modbus.exceptions import NeoPoolError
 from neopool_modbus.registers import (
     MAX_RELAY_GPIO,
+    TIMER_BLOCKS,
     find_corrupted_gpio_registers,
     is_valid_relay_gpio,
 )
@@ -32,12 +34,19 @@ from .const import (
 
 _LOGGER = logging.getLogger(__name__)
 
-# Auxiliary relay timer blocks keyed by their enabling option.
-_AUX_TIMER_BLOCKS: dict[str, str] = {
-    CONF_USE_AUX1: "relay_aux1",
-    CONF_USE_AUX2: "relay_aux2",
-    CONF_USE_AUX3: "relay_aux3",
-    CONF_USE_AUX4: "relay_aux4",
+_FILT_TIMERS = ("filtration1", "filtration2", "filtration3")
+
+# Config option gating each aux and light timer block.
+_TIMER_OPTIONS: dict[str, str] = {
+    "relay_aux1": CONF_USE_AUX1,
+    "relay_aux1b": CONF_USE_AUX1,
+    "relay_aux2": CONF_USE_AUX2,
+    "relay_aux2b": CONF_USE_AUX2,
+    "relay_aux3": CONF_USE_AUX3,
+    "relay_aux3b": CONF_USE_AUX3,
+    "relay_aux4": CONF_USE_AUX4,
+    "relay_aux4b": CONF_USE_AUX4,
+    "relay_light": CONF_USE_LIGHT,
 }
 
 
@@ -69,6 +78,11 @@ class NeoPoolCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         self._follow_up_unsub: CALLBACK_TYPE | None = None
         # Serializes masked read-modify-write across siblings sharing a register.
         self.masked_write_lock = asyncio.Lock()
+        # One lock per timer block serializes the library's read-modify-write
+        # across the block's start/stop sibling entities, which share a register.
+        self._timer_write_locks: defaultdict[str, asyncio.Lock] = defaultdict(
+            asyncio.Lock
+        )
 
     def request_refresh_with_followup(
         self, delay: float = FOLLOW_UP_REFRESH_DELAY
@@ -93,6 +107,10 @@ class NeoPoolCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             self._follow_up_unsub()
             self._follow_up_unsub = None
 
+    def timer_write_lock(self, block: str) -> asyncio.Lock:
+        """Return the lock serializing sibling writes to one timer block."""
+        return self._timer_write_locks[block]
+
     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)
@@ -133,31 +151,41 @@ class NeoPoolCoordinator(DataUpdateCoordinator[dict[str, Any]]):
             ir.async_delete_issue(self.hass, DOMAIN, "corrupted_gpio")
 
     def _get_enabled_timers(self, data: dict[str, Any]) -> list[str]:
-        """Return the list of timer block names to poll each cycle.
+        """Return the timer block names to poll.
 
-        The light timer is polled only when its option is enabled and the
-        lighting GPIO is valid. Auxiliary relay timers are polled per enabled
-        option; the aux switches read relay_aux*_enable as a manual-mode guard.
+        Base aux and light blocks poll on their config option. The second aux
+        subtimer and filtration blocks additionally require an active context
+        (the block name registered by an enabled time entity).
         """
+        options = self.config_entry.options
+        active = {ctx for ctx in self.async_contexts() if isinstance(ctx, str)}
         enabled: list[str] = []
-        if self.config_entry.options.get(CONF_USE_LIGHT, False) and is_valid_relay_gpio(
-            data.get("MBF_PAR_LIGHTING_GPIO", 0) or 0
-        ):
-            enabled.append("relay_light")
-        enabled.extend(
-            block
-            for option, block in _AUX_TIMER_BLOCKS.items()
-            if self.config_entry.options.get(option, False)
-        )
+        for key in TIMER_BLOCKS:
+            option_key = _TIMER_OPTIONS.get(key)
+            if option_key is None:
+                # Filtration timers gate on context below, not an option.
+                continue
+            if not options.get(option_key, False):
+                continue
+            # The b subtimer (time only) also needs an active context.
+            if key.endswith("b") and key not in active:
+                continue
+            # Light GPIO invalid: the light entity gates the same, so
+            # relay_light_enable has no consumer.
+            if key == "relay_light" and not is_valid_relay_gpio(
+                data.get("MBF_PAR_LIGHTING_GPIO", 0) or 0
+            ):
+                continue
+            enabled.append(key)
+        enabled += [ft for ft in _FILT_TIMERS if ft in active]
         return enabled
 
     async def _read_timers_into_data(self, data: dict[str, Any]) -> None:
         """Read every enabled timer block and merge derived fields into data.
 
-        Only the ``<timer>_enable`` field is exposed: it is the sole timer
-        attribute consumed by the light platform (as a manual-mode guard).
-        Further derived keys will be added by follow-up platform PRs that
-        consume them.
+        Exposes the enable flag (the light platform's manual-mode guard) and
+        the start/stop endpoints consumed by the time platform. Further derived
+        keys will be added by follow-up platform PRs that consume them.
         """
         enabled = self._get_enabled_timers(data)
         if not enabled:
@@ -165,6 +193,8 @@ class NeoPoolCoordinator(DataUpdateCoordinator[dict[str, Any]]):
         timers = await self.client.read_all_timers(enabled_timers=enabled)
         for t_name, t in timers.items():
             data[f"{t_name}_enable"] = t["enable"]
+            data[f"{t_name}_start"] = t["on"]  # seconds since midnight
+            data[f"{t_name}_stop"] = t.get("stop")
 
     @override
     async def _async_update_data(self) -> dict[str, Any]:

homeassistant/components/neopool/icons.json

@@ -146,6 +146,26 @@
           "on": "mdi:lightbulb-fluorescent-tube"
         }
       }
+    },
+    "time": {
+      "filtration_start": {
+        "default": "mdi:clock-start"
+      },
+      "filtration_stop": {
+        "default": "mdi:clock-end"
+      },
+      "relay_aux_start": {
+        "default": "mdi:clock-start"
+      },
+      "relay_aux_stop": {
+        "default": "mdi:clock-end"
+      },
+      "relay_light_start": {
+        "default": "mdi:clock-start"
+      },
+      "relay_light_stop": {
+        "default": "mdi:clock-end"
+      }
     }
   },
   "services": {

homeassistant/components/neopool/strings.json

@@ -314,6 +314,26 @@
       "uv_mode": {
         "name": "UV mode"
       }
+    },
+    "time": {
+      "filtration_start": {
+        "name": "Timer {number} - filtration start"
+      },
+      "filtration_stop": {
+        "name": "Timer {number} - filtration stop"
+      },
+      "relay_aux_start": {
+        "name": "Timer Aux {number} ({subtimer}) - start"
+      },
+      "relay_aux_stop": {
+        "name": "Timer Aux {number} ({subtimer}) - stop"
+      },
+      "relay_light_start": {
+        "name": "Light timer - start"
+      },
+      "relay_light_stop": {
+        "name": "Light timer - stop"
+      }
     }
   },
   "exceptions": {

homeassistant/components/neopool/time.py

@@ -0,0 +1,410 @@
+"""Time platform for the NeoPool integration."""
+
+import asyncio
+from collections.abc import Callable, Mapping
+from contextlib import suppress
+from dataclasses import dataclass
+from datetime import datetime, time as dt_time, timedelta
+from typing import Any, Literal, override
+
+from neopool_modbus.exceptions import NeoPoolError
+from neopool_modbus.registers import is_valid_relay_gpio
+
+from homeassistant.components.time import TimeEntity, TimeEntityDescription
+from homeassistant.const import EntityCategory
+from homeassistant.core import CALLBACK_TYPE, HomeAssistant, callback
+from homeassistant.exceptions import HomeAssistantError
+from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
+from homeassistant.helpers.event import async_call_later
+
+from .const import (
+    CONF_USE_AUX1,
+    CONF_USE_AUX2,
+    CONF_USE_AUX3,
+    CONF_USE_AUX4,
+    CONF_USE_LIGHT,
+    DOMAIN,
+)
+from .coordinator import NeoPoolConfigEntry, NeoPoolCoordinator
+from .entity import NeoPoolEntity
+
+# The platform coalesces rapid writes per entity via a debounce timer. Each
+# start/stop entity flushes its own endpoint independently, so a platform
+# semaphore would only add latency between independent UI interactions.
+PARALLEL_UPDATES = 0
+
+# Wait for editing to settle so only the final value hits the device's EEPROM.
+WRITE_DELAY = timedelta(seconds=3)
+
+
+@dataclass(frozen=True, kw_only=True)
+class NeoPoolTimeEntityDescription(TimeEntityDescription):
+    """NeoPool time entity description."""
+
+    timer_block: str
+    timer_field: Literal["start", "stop"]
+    supported_fn: Callable[[dict[str, Any], Mapping[str, Any]], bool] | None = None
+    translation_placeholders: dict[str, str] | None = None
+
+
+def _option_supported(
+    opt_flag: str,
+) -> Callable[[dict[str, Any], Mapping[str, Any]], bool]:
+    """Return a supported_fn gating an entity on the given option flag."""
+    return lambda _data, opts: bool(opts.get(opt_flag))
+
+
+def _light_supported(data: dict[str, Any], opts: Mapping[str, Any]) -> bool:
+    """Gate the light timer on its option and a valid lighting GPIO.
+
+    The coordinator skips the relay_light block when the GPIO is invalid.
+    """
+    return bool(opts.get(CONF_USE_LIGHT)) and is_valid_relay_gpio(
+        data.get("MBF_PAR_LIGHTING_GPIO", 0) or 0
+    )
+
+
+# Filtration timers exist on every device; aux and light timers gate on options.
+_TIMER_BLOCKS: tuple[tuple[str, str | None, bool], ...] = (
+    ("filtration1", None, True),
+    ("filtration2", None, False),
+    ("filtration3", None, False),
+    ("relay_aux1", CONF_USE_AUX1, True),
+    ("relay_aux1b", CONF_USE_AUX1, False),
+    ("relay_aux2", CONF_USE_AUX2, True),
+    ("relay_aux2b", CONF_USE_AUX2, False),
+    ("relay_aux3", CONF_USE_AUX3, True),
+    ("relay_aux3b", CONF_USE_AUX3, False),
+    ("relay_aux4", CONF_USE_AUX4, True),
+    ("relay_aux4b", CONF_USE_AUX4, False),
+    ("relay_light", CONF_USE_LIGHT, True),
+)
+
+
+def _build_descriptions() -> dict[str, NeoPoolTimeEntityDescription]:
+    """Build the timer start/stop descriptions."""
+    out: dict[str, NeoPoolTimeEntityDescription] = {}
+    for block, opt_flag, enabled_default in _TIMER_BLOCKS:
+        for field in ("start", "stop"):
+            key = f"{block}_{field}"
+            # Filtration and aux blocks share one translation per field with
+            # number placeholders; other blocks keep their own key.
+            translation_key = key
+            placeholders: dict[str, str] | None = None
+            if block.startswith("filtration"):
+                translation_key = f"filtration_{field}"
+                placeholders = {"number": block.removeprefix("filtration")}
+            elif block.startswith("relay_aux"):
+                translation_key = f"relay_aux_{field}"
+                digits = block.removeprefix("relay_aux")
+                placeholders = {
+                    "number": digits.rstrip("b"),
+                    "subtimer": "2" if digits.endswith("b") else "1",
+                }
+            if block == "relay_light":
+                supported_fn: (
+                    Callable[[dict[str, Any], Mapping[str, Any]], bool] | None
+                ) = _light_supported
+            elif opt_flag is not None:
+                supported_fn = _option_supported(opt_flag)
+            else:
+                supported_fn = None
+            out[key] = NeoPoolTimeEntityDescription(
+                key=key,
+                translation_key=translation_key,
+                translation_placeholders=placeholders,
+                entity_category=EntityCategory.CONFIG,
+                entity_registry_enabled_default=enabled_default,
+                timer_block=block,
+                timer_field=field,
+                supported_fn=supported_fn,
+            )
+    return out
+
+
+TIME_DESCRIPTIONS: dict[str, NeoPoolTimeEntityDescription] = _build_descriptions()
+
+
+async def async_setup_entry(
+    hass: HomeAssistant,
+    entry: NeoPoolConfigEntry,
+    async_add_entities: AddConfigEntryEntitiesCallback,
+) -> None:
+    """Set up NeoPool time entities from a config entry."""
+    coordinator = entry.runtime_data
+
+    async_add_entities(
+        NeoPoolTime(coordinator, key, desc)
+        for key, desc in TIME_DESCRIPTIONS.items()
+        if desc.supported_fn is None
+        or desc.supported_fn(coordinator.data, entry.options)
+    )
+
+
+class NeoPoolTime(NeoPoolEntity, TimeEntity):
+    """NeoPool timer start/stop time entity."""
+
+    entity_description: NeoPoolTimeEntityDescription
+
+    def __init__(
+        self,
+        coordinator: NeoPoolCoordinator,
+        key: str,
+        description: NeoPoolTimeEntityDescription,
+    ) -> None:
+        """Initialize the entity."""
+        # Filtration and second-aux-subtimer blocks poll only while an entity
+        # is enabled, so register the block as coordinator context. Base aux and
+        # light poll on their option flag and need none.
+        block = description.timer_block
+        context = (
+            block if (description.supported_fn is None or block.endswith("b")) else None
+        )
+        super().__init__(coordinator, context=context)
+        self.entity_description = description
+        self._key = key
+        if description.translation_placeholders is not None:
+            self._attr_translation_placeholders = description.translation_placeholders
+        self._attr_unique_id = (
+            f"{self.coordinator.config_entry.unique_id}_{key.lower()}"
+        )
+
+        self._write_unsub: CALLBACK_TYPE | None = None
+        # Optimistic value pending a write, held as seconds-since-midnight.
+        self._pending_value: int | None = None
+        # Bumped per set_value; a flush clears only the value it queued.
+        self._pending_token = 0
+        self._write_future: asyncio.Future[Exception | None] | None = None
+        self._flush_lock = asyncio.Lock()
+        self._flush_tasks: set[asyncio.Task[None]] = set()
+        self._removing = False
+
+    def _decode_raw(self) -> dt_time | None:
+        """Decode the coordinator-data seconds into HH:MM:SS."""
+        seconds = self.coordinator.data.get(self._key)
+        if seconds is None:
+            return None
+        try:
+            seconds = int(seconds) % 86400
+        except TypeError, ValueError:  # pragma: no cover
+            return None
+        return dt_time(
+            hour=seconds // 3600,
+            minute=(seconds % 3600) // 60,
+            second=seconds % 60,
+        )
+
+    @property
+    @override
+    def native_value(self) -> dt_time | None:
+        """Return the optimistic pending time, else the decoded register."""
+        if self._pending_value is not None:
+            seconds = self._pending_value % 86400
+            return dt_time(
+                hour=seconds // 3600,
+                minute=(seconds % 3600) // 60,
+                second=seconds % 60,
+            )
+        return self._decode_raw()
+
+    @override
+    async def async_added_to_hass(self) -> None:
+        """Clear transient write state, in case this entity is re-added.
+
+        An entity-ID change re-adds the same object, leaving _removing set and a
+        cancelled pending value behind; reset both so later flushes do not abort.
+        """
+        self._removing = False
+        self._pending_value = None
+        await super().async_added_to_hass()
+
+    @override
+    async def async_will_remove_from_hass(self) -> None:
+        """Cancel a pending write when removed, and stop any in-flight one."""
+        self._removing = True
+        self._cancel_pending_write()
+        if self._write_future is not None and not self._write_future.done():
+            # Awaiting callers treat cancellation as a clean exit.
+            self._write_future.cancel()
+        # Cancel and await every in-flight flush so no device call outlives
+        # removal and races the client close. Two set_value calls spaced beyond
+        # WRITE_DELAY can overlap, so more than one task may be active.
+        for task in list(self._flush_tasks):
+            task.cancel()
+        for task in list(self._flush_tasks):
+            with suppress(asyncio.CancelledError):
+                await task
+        await super().async_will_remove_from_hass()
+
+    @callback
+    def _cancel_pending_write(self) -> None:
+        """Cancel a scheduled write, if any."""
+        if self._write_unsub is not None:
+            self._write_unsub()
+            self._write_unsub = None
+
+    @override
+    async def async_set_value(self, value: dt_time) -> None:
+        """Apply optimistically, then debounce-write to the device.
+
+        The write is debounced so rapid edits settle into a single EEPROM
+        cycle. Callers in the same window await one shared future the coalesced
+        write resolves, so a blocking service call still sees the outcome.
+        """
+        self._pending_value = value.hour * 3600 + value.minute * 60 + value.second
+        # A fresh token per set_value lets a flush clear exactly the value it
+        # queued, not a newer batch's identical one.
+        self._pending_token += 1
+        self.async_write_ha_state()
+        self._cancel_pending_write()
+        if self._write_future is None or self._write_future.done():
+            self._write_future = self.hass.loop.create_future()
+        future = self._write_future
+        self._write_unsub = async_call_later(
+            self.hass, WRITE_DELAY, self._schedule_flush
+        )
+        try:
+            # Shield so cancelling one caller does not cancel the batch. The
+            # write carries its outcome as the future's result (None on success,
+            # else the error to re-raise) rather than via set_exception, which a
+            # cancelled caller's shield logger would report as unretrieved.
+            outcome = await asyncio.shield(future)
+        except asyncio.CancelledError:
+            if self._removing:
+                return
+            raise
+        if outcome is not None:
+            raise outcome
+
+    @callback
+    def _schedule_flush(self, _now: datetime) -> None:
+        """Run the debounced write as a tracked task so removal can await it."""
+        self._write_unsub = None
+        # Detach this batch synchronously so a set_value racing _async_flush
+        # starts a fresh future and timer instead of reusing this one.
+        # _pending_value stays put to back the optimistic value.
+        future = self._write_future
+        self._write_future = None
+        token = self._pending_token
+        pending = self._pending_value
+        task = self.coordinator.config_entry.async_create_background_task(
+            self.hass,
+            self._async_flush(future, pending, token),
+            name=f"{self._attr_unique_id}_flush",
+        )
+        # Track every in-flight flush so removal can cancel and await them all;
+        # a set_value spaced beyond WRITE_DELAY can start a second task while an
+        # earlier one is still writing.
+        self._flush_tasks.add(task)
+        task.add_done_callback(self._flush_tasks.discard)
+
+    async def _async_flush(
+        self,
+        future: asyncio.Future[Exception | None] | None,
+        pending: int | None,
+        token: int,
+    ) -> None:
+        """Write the settled value, resolving the awaited coalesce future."""
+        resolved = False
+        try:
+            if pending is None:  # pragma: no cover - timer fires only when queued
+                return
+            async with self._flush_lock:
+                if self._abort_if_removing(future):
+                    resolved = True
+                    return
+                block = self.entity_description.timer_block
+                # The library does the read-modify-write, so pass only this
+                # entity's endpoint: start -> on, stop -> stop.
+                lib_key = (
+                    "on" if self.entity_description.timer_field == "start" else "stop"
+                )
+                # Skip the EEPROM cycle if the device already holds this value.
+                if (current := self._decode_raw()) is not None and (
+                    current.hour * 3600 + current.minute * 60 + current.second
+                    == pending % 86400
+                ):
+                    self._clear_pending_if_current(token)
+                    if future is not None and not future.done():
+                        future.set_result(None)
+                    resolved = True
+                    return
+                try:
+                    # Siblings (start/stop) share the block's register set, so
+                    # serialize the library's read-modify-write per block.
+                    async with self.coordinator.timer_write_lock(block):
+                        await self.coordinator.client.write_timer(
+                            block, {lib_key: pending}
+                        )
+                except (NeoPoolError, OSError, TimeoutError) as err:
+                    self._report_write_failure(
+                        future,
+                        token,
+                        HomeAssistantError(
+                            translation_domain=DOMAIN,
+                            translation_key="modbus_communication_error",
+                            translation_placeholders={"error": str(err)},
+                        ),
+                    )
+                    resolved = True
+                    return
+                except Exception as err:  # noqa: BLE001
+                    # Surface unexpected errors unchanged, not as a comm error.
+                    self._report_write_failure(future, token, err)
+                    resolved = True
+                    return
+                if self._abort_if_removing(future):  # pragma: no cover
+                    # Defensive: removal cancels tracked tasks before they write.
+                    resolved = True
+                    return
+                try:
+                    # Merge before clearing, else the stale register briefly
+                    # surfaces as a rollback.
+                    self.coordinator.async_set_updated_data(
+                        {**self.coordinator.data, self._key: pending}
+                    )
+                    self._clear_pending_if_current(token)
+                    self.coordinator.request_refresh_with_followup()
+                except Exception as err:  # noqa: BLE001
+                    # Write succeeded; surface the merge error unchanged.
+                    self._report_write_failure(future, token, err)
+                    resolved = True
+                    return
+                if future is not None and not future.done():
+                    future.set_result(None)
+                resolved = True
+        finally:
+            if not resolved and future is not None and not future.done():
+                future.cancel()  # pragma: no cover - task cancel is non-deterministic
+
+    @callback
+    def _abort_if_removing(
+        self, future: asyncio.Future[Exception | None] | None
+    ) -> bool:
+        """Skip the write when removed, releasing the detached future cleanly."""
+        if not self._removing:
+            return False
+        if future is not None and not future.done():
+            future.cancel()
+        return True
+
+    @callback
+    def _report_write_failure(
+        self,
+        future: asyncio.Future[Exception | None] | None,
+        batch_token: int,
+        exc: Exception,
+    ) -> None:
+        """Roll the optimistic value back and fail the awaiting caller."""
+        self._clear_pending_if_current(batch_token)
+        if future is not None and not future.done():
+            # Carry the error as the result, not via set_exception; callers
+            # re-raise it after the shield returns.
+            future.set_result(exc)
+
+    @callback
+    def _clear_pending_if_current(self, batch_token: int) -> None:
+        """Drop the optimistic value unless a newer set_value replaced it."""
+        if self._pending_token == batch_token:
+            self._pending_value = None
+        self.async_write_ha_state()

tests/components/neopool/conftest.py

@@ -118,10 +118,18 @@ def _timer_block(enable: int = 4) -> dict[str, Any]:
 
 
 MOCK_TIMER_BLOCKS: dict[str, dict[str, Any]] = {
+    "filtration1": _timer_block(),
+    "filtration2": _timer_block(),
+    "filtration3": _timer_block(),
     "relay_aux1": _timer_block(),
+    "relay_aux1b": _timer_block(),
     "relay_aux2": _timer_block(),
+    "relay_aux2b": _timer_block(),
     "relay_aux3": _timer_block(),
+    "relay_aux3b": _timer_block(),
     "relay_aux4": _timer_block(),
+    "relay_aux4b": _timer_block(),
+    "relay_light": _timer_block(),
 }
 
 
@@ -209,6 +217,31 @@ def mock_config_entry_switch() -> MockConfigEntry:
     )
 
 
+@pytest.fixture
+def mock_config_entry_timers() -> MockConfigEntry:
+    """Return a config entry with the aux and light timer options enabled."""
+    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_AUX1: True,
+            CONF_USE_AUX2: True,
+            CONF_USE_AUX3: True,
+            CONF_USE_AUX4: True,
+        },
+    )
+
+
 @pytest.fixture
 def mock_config_entry_number() -> MockConfigEntry:
     """Return a config entry with the options the number platform gates on."""

tests/components/neopool/snapshots/test_diagnostics.ambr

@@ -86,6 +86,9 @@
         'Pool Cover': 0,
         'Redox measurement active': True,
         'Redox measurement module detected': True,
+        'filtration1_enable': 4,
+        'filtration1_start': 0,
+        'filtration1_stop': None,
         'filtration_mode': 'manual',
         'filtration_speed_state': 'off',
         'pH Acid Pump': False,

tests/components/neopool/test_coordinator.py

@@ -0,0 +1,387 @@
+"""Tests for the NeoPool coordinator."""
+
+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 (
+    CONF_MODBUS_FRAMER,
+    CONF_UNIT_ID,
+    CONF_USE_AUX1,
+    CURRENT_VERSION,
+    DOMAIN,
+)
+from homeassistant.config_entries import ConfigEntryState
+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, _read_all_timers
+
+from tests.common import MockConfigEntry, async_fire_time_changed
+
+
+@pytest.mark.usefixtures("mock_neopool_client")
+async def test_update_data_populates_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_by_identifier(
+        (DOMAIN, MOCK_SERIAL), mock_config_entry.entry_id
+    )
+    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:
+    """A failure after at least one good read raises UpdateFailed (not ConfigEntryNotReady)."""
+    await setup_integration(hass, mock_config_entry)
+    coordinator = mock_config_entry.runtime_data
+    assert coordinator.last_update_success is True
+
+    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 coordinator.last_update_success is False
+    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
+
+
+async def test_corrupt_gpio_self_heals_on_next_clean_read(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_neopool_client: MagicMock,
+    issue_registry: ir.IssueRegistry,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """The corrupted_gpio issue clears once a subsequent poll reads clean values."""
+    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 issue_registry.async_get_issue(DOMAIN, "corrupted_gpio") is not None
+
+    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
+
+
+@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:
+    """Stale issue from a previous HA session clears on first 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,
+    freezer: FrozenDateTimeFactory,
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    """ERROR log fires only when the set of corrupted register keys changes."""
+    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)
+
+    initial_errors = sum(
+        1 for r in caplog.records if "Corrupted GPIO register" in r.getMessage()
+    )
+    assert initial_errors == 1
+
+    freezer.tick(timedelta(seconds=60))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    total_errors = sum(
+        1 for r in caplog.records if "Corrupted GPIO register" in r.getMessage()
+    )
+    assert total_errors == 1
+
+
+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"]
+
+
+async def test_follow_up_refresh_callback_runs(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_neopool_client: MagicMock,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """request_refresh_with_followup schedules a refresh that fires after the delay."""
+    await setup_integration(hass, mock_config_entry)
+    coordinator = mock_config_entry.runtime_data
+
+    initial_count = mock_neopool_client.async_read_all.await_count
+    coordinator.request_refresh_with_followup(delay=0.1)
+    freezer.tick(timedelta(seconds=0.2))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+    assert mock_neopool_client.async_read_all.await_count > initial_count
+
+
+async def test_timer_block_data_merged_into_coordinator(
+    hass: HomeAssistant,
+    mock_config_entry: MockConfigEntry,
+    mock_neopool_client: MagicMock,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """When read_all_timers returns timer blocks, the per-block fields land in data."""
+
+    def _timers(
+        enabled_timers: list[str] | None = None, **_kwargs: object
+    ) -> dict[str, dict[str, object]]:
+        return {
+            "filtration1": {
+                "enable": 1,
+                "on": 8 * 3600,
+                "interval": 4 * 3600,
+                "stop": 12 * 3600,
+                "period": 86400,
+                "countdown": 3600,
+            },
+            "filtration2": {
+                "enable": 0,
+                "on": None,
+                "interval": None,
+                "stop": None,
+                "period": None,
+                "countdown": 0,
+            },
+        }
+
+    mock_neopool_client.read_all_timers.side_effect = _timers
+    await setup_integration(hass, mock_config_entry)
+    # filtration1 start/stop register their context after setup; poll again so
+    # the coordinator reads their block.
+    freezer.tick(timedelta(seconds=60))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    coordinator = mock_config_entry.runtime_data
+
+    assert coordinator.data["filtration1_enable"] == 1
+    assert coordinator.data["filtration1_start"] == 8 * 3600
+    assert coordinator.data["filtration1_stop"] == 12 * 3600
+    assert coordinator.data["filtration2_stop"] is None
+
+
+def _capture_timer_calls(mock_neopool_client: MagicMock) -> list[tuple]:
+    """Record the enabled_timers list per read_all_timers call."""
+    calls: list[tuple] = []
+
+    def _capture(
+        enabled_timers: list[str] | None = None, **kwargs: object
+    ) -> dict[str, dict[str, object]]:
+        calls.append((enabled_timers,))
+        return _read_all_timers(enabled_timers)
+
+    mock_neopool_client.read_all_timers.side_effect = _capture
+    return calls
+
+
+async def _poll_once_more(
+    hass: HomeAssistant,
+    coordinator: object,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Trigger a poll after entity listeners have registered their contexts."""
+    freezer.tick(timedelta(seconds=60))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+
+
+async def test_no_filtration_polled_when_all_entities_disabled(
+    hass: HomeAssistant,
+    mock_neopool_client: MagicMock,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """With filtration2/3 time entities disabled, only filtration1 polls.
+
+    filtration1 start/stop are registry-enabled by default; filtration2/3 are
+    not, so their blocks must stay out of the read.
+    """
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        title="Pool",
+        unique_id="neopool_gate_default",
+        version=CURRENT_VERSION,
+        data={
+            "host": "192.0.2.20",
+            "port": 502,
+            "name": "Pool",
+            CONF_UNIT_ID: 1,
+            CONF_MODBUS_FRAMER: "tcp",
+        },
+        options={CONF_MODBUS_FRAMER: "tcp"},
+    )
+    calls = _capture_timer_calls(mock_neopool_client)
+    await setup_integration(hass, entry)
+    await _poll_once_more(hass, entry.runtime_data, freezer)
+
+    enabled_timers = calls[-1][0]
+    assert enabled_timers is not None
+    assert "filtration1" in enabled_timers
+    assert "filtration2" not in enabled_timers
+    assert "filtration3" not in enabled_timers
+
+
+async def test_aux_base_polls_but_b_subtimer_gated_when_disabled(
+    hass: HomeAssistant,
+    mock_neopool_client: MagicMock,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """With use_aux1 on, the base block polls but the second subtimer does not.
+
+    relay_aux1 stays option-gated (the aux switch needs its enable state); the
+    relay_aux1b start/stop entities are registry-disabled by default, so they
+    never register a context and the block stays out of the read.
+    """
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        title="Pool",
+        unique_id="neopool_gate_aux_default",
+        version=CURRENT_VERSION,
+        data={
+            "host": "192.0.2.22",
+            "port": 502,
+            "name": "Pool",
+            CONF_UNIT_ID: 1,
+            CONF_MODBUS_FRAMER: "tcp",
+        },
+        options={CONF_MODBUS_FRAMER: "tcp", CONF_USE_AUX1: True},
+    )
+    calls = _capture_timer_calls(mock_neopool_client)
+    await setup_integration(hass, entry)
+    await _poll_once_more(hass, entry.runtime_data, freezer)
+
+    enabled_timers = calls[-1][0]
+    assert enabled_timers is not None
+    assert "relay_aux1" in enabled_timers
+    assert "relay_aux1b" not in enabled_timers
+
+
+@pytest.mark.usefixtures("entity_registry_enabled_by_default")
+async def test_aux_b_subtimer_polls_when_enabled(
+    hass: HomeAssistant,
+    mock_neopool_client: MagicMock,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Enabling the second aux subtimer's entities starts polling its block.
+
+    With every entity enabled, relay_aux1b start/stop register their block as
+    an update context, so it joins the read alongside the base block.
+    """
+    entry = MockConfigEntry(
+        domain=DOMAIN,
+        title="Pool",
+        unique_id="neopool_gate_aux_b",
+        version=CURRENT_VERSION,
+        data={
+            "host": "192.0.2.23",
+            "port": 502,
+            "name": "Pool",
+            CONF_UNIT_ID: 1,
+            CONF_MODBUS_FRAMER: "tcp",
+        },
+        options={CONF_MODBUS_FRAMER: "tcp", CONF_USE_AUX1: True},
+    )
+    calls = _capture_timer_calls(mock_neopool_client)
+    await setup_integration(hass, entry)
+    await _poll_once_more(hass, entry.runtime_data, freezer)
+
+    enabled_timers = calls[-1][0]
+    assert enabled_timers is not None
+    assert "relay_aux1" in enabled_timers
+    assert "relay_aux1b" in enabled_timers

tests/components/neopool/test_light.py

@@ -128,30 +128,52 @@ async def test_light_is_on_reflects_relay_state(
 
 
 @pytest.mark.parametrize(
-    "relay_data",
+    "light_enable",
     [
-        pytest.param({"relay_light_enable": TimerRelayMode.ENABLED}, id="auto"),
-        pytest.param({}, id="missing"),
-        pytest.param({"relay_light_enable": 0}, id="disabled"),
-        pytest.param({"relay_light_enable": 2}, id="unknown-state"),
+        pytest.param(TimerRelayMode.ENABLED, id="auto"),
+        pytest.param(None, id="missing"),
+        pytest.param(0, id="disabled"),
+        pytest.param(2, id="unknown-state"),
     ],
 )
 async def test_light_refuses_when_not_in_manual_mode(
     hass: HomeAssistant,
     mock_config_entry_light: MockConfigEntry,
     mock_neopool_client: MagicMock,
-    freezer: FrozenDateTimeFactory,
-    relay_data: dict[str, int],
+    light_enable: int | None,
 ) -> None:
     """Turn on/off is rejected while the relay is not in a manual mode."""
+
+    # The coordinator derives relay_light_enable from the timer block, so drive
+    # the guard through read_all_timers, not the async_read_all payload. Set the
+    # side_effect before setup so the very first poll reflects the case; a None
+    # case omits the block entirely, leaving relay_light_enable absent.
+    def _timers(
+        enabled_timers: list[str] | None = None, **_kwargs: object
+    ) -> dict[str, dict[str, object]]:
+        if light_enable is None:
+            return {}
+        return {
+            "relay_light": {
+                "enable": light_enable,
+                "on": 0,
+                "interval": 0,
+                "period": 0,
+                "countdown": 0,
+                "stop": None,
+            }
+        }
+
+    # Drop the autouse fixture's seeded manual enable so the timer read is the
+    # only source of relay_light_enable; the missing case must leave it absent.
+    mock_neopool_client.async_read_all.return_value = {
+        **MOCK_POOL_DATA,
+        "Pool Light": False,
+    }
+    mock_neopool_client.read_all_timers.side_effect = _timers
     await setup_integration(hass, mock_config_entry_light)
     entity_id = _light_entity_id(hass, mock_config_entry_light)
 
-    mock_neopool_client.async_read_all.return_value = {**MOCK_POOL_DATA, **relay_data}
-    freezer.tick(timedelta(seconds=60))
-    async_fire_time_changed(hass)
-    await hass.async_block_till_done()
-
     mock_neopool_client.async_set_relay_state.reset_mock()
     with pytest.raises(ServiceValidationError):
         await _turn_on(hass, entity_id)