Merge pull request #2200 from rosenrot00/fix/atomic-poll-snapshots

Publish polling data as consistent snapshots

Author: wills106

custom_components/solax_modbus/__init__.py

@@ -181,6 +181,7 @@ def empty_hub_device_group_lambda() -> SimpleNamespace:
         holdingBlocks={},
         readPreparation=None,  # function to call before read group
         readFollowUp=None,  # function to call after read group
+        publish_updates=False,
     )
 
 
@@ -543,6 +544,7 @@ def __init__(
             # Fallback dummy client for unrecognized interface types
             self._client = SimpleNamespace(connected=False, comm_params=SimpleNamespace(host="", port=""))
         self._lock = asyncio.Lock()
+        self._poll_data_lock = asyncio.Lock()
         self._name: str = name
         # following call will modify and extend client in case old modbus API needs to be used
         _LOGGER.debug(f"{name}: using pymodbus version {pymodbus_version_info()}")
@@ -1053,9 +1055,10 @@ async def _refresh_interval_group_once(self, interval_group: Any, bypass_slowdow
                     if self.slowdown > 1:
                         _LOGGER.debug(f"{self._name}: communication restored, resuming normal speed after slowdown")
                     self.slowdown = 1
-                    for sensor in group.sensors:
-                        sensor.modbus_data_updated()
-                    updated_sensors += len(group.sensors)
+                    if getattr(group, "publish_updates", True):
+                        for sensor in group.sensors:
+                            sensor.modbus_data_updated()
+                        updated_sensors += len(group.sensors)
                 else:
                     if self.slowdown <= 1:
                         _LOGGER.debug(f"{self._name}: modbus group read failed - assuming sleep mode - slowing down by factor 10")
@@ -1535,8 +1538,10 @@ async def async_write_registers_multi(self, unit: int, address: int, payload: li
 
     async def async_read_modbus_data(self, group: Any) -> bool:
         res = True
+        group.publish_updates = False
         try:
-            res = await self.async_read_modbus_registers_all(group)
+            async with self._poll_data_lock:
+                res = await self.async_read_modbus_registers_all(group)
         except ConnectionException as ex:
             _LOGGER.error(f"Reading data failed! Inverter is offline. {ex}")
             res = False
@@ -1677,7 +1682,7 @@ def treat_address(self, data: dict[str, Any], regs: list[int], idx: int, descr:
         # if (descr.sleepmode != SLEEPMODE_LASTAWAKE) or self.awakeplugin(self.data): self.data[descr.key] = return_value
         if (
             (self.tmpdata_expiry.get(descr.key, 0) == 0)
-            and ((descr.sleepmode != SLEEPMODE_LASTAWAKE) or self.plugin.isAwake(self.data))
+            and ((descr.sleepmode != SLEEPMODE_LASTAWAKE) or self.plugin.isAwake(data))
             and (self.localsLoaded or not descr.read_scale_exceptions)  # ignore as long as read scale is not adapted; may delay real startup a bit
         ):
             data[descr.key] = return_value  # case prevent_update number
@@ -1767,16 +1772,34 @@ async def async_read_modbus_block(self, data: dict[str, Any], block: Any, typ: s
                     )
                 return False
 
+    def _commit_poll_snapshot(self, previous_data: dict[str, Any], new_data: dict[str, Any]) -> None:
+        """Commit polling changes without replacing the shared data dictionary."""
+        missing = object()
+
+        for key in previous_data.keys() - new_data.keys():
+            if self.data.get(key, missing) == previous_data[key]:
+                self.data.pop(key, None)
+
+        for key, value in new_data.items():
+            previous_value = previous_data.get(key, missing)
+            if previous_value is not missing and value == previous_value:
+                continue
+
+            current_value = self.data.get(key, missing)
+            if current_value is missing or current_value == previous_value:
+                self.data[key] = value
+
     async def async_read_modbus_registers_all(self, group: Any) -> bool:
+        group.publish_updates = False
         if group.readPreparation is not None:
             if not await group.readPreparation(self.data):
                 _LOGGER.info(f"{self._name}: device group read cancel")
                 return True
         else:
             _LOGGER.debug(f"{self._name}: device group inverter")
 
-        # data = {"_repeatUntil": self.data["_repeatUntil"]} # remove for issue #1440 but then does not recognize comm errors
-        data = self.data  # is an alias, not a copy (issue #1440)
+        previous_data = self.data.copy()
+        data = previous_data.copy()
         res = True
         for block in group.holdingBlocks:
             _LOGGER.debug(f"{self._name}: ** trying to read holding block 0x{block.start:x} previous res:{res}")
@@ -1789,32 +1812,44 @@ async def async_read_modbus_registers_all(self, group: Any) -> bool:
             res = res and block_res
             _LOGGER.debug(f"{self._name}: input block 0x{block.start:x} read done; new res: {res}")
 
+        local_callback_needed = self.localsUpdated
         if self.localsUpdated:
             await self._hass.async_add_executor_job(self.saveLocalData)
             self.plugin.localDataCallback(self)
         if not self.localsLoaded:
             await self._hass.async_add_executor_job(self.loadLocalData)
-        for key, descr in list(self.computedSensors.items()):
-            try:
-                data[key] = descr.value_function(0, descr, data)
-            except Exception as ex:
-                _LOGGER.debug(f"{self._name}: cannot compute value for {key}: {ex}")
-                continue
-            sens = self.sensorEntities.get(key)
-            _LOGGER.debug(f"{self._name}: quickly updating state for computed sensor {sens} {key} {data.get(descr.key)} ")
-            if sens and (not descr.internal):
-                try:
-                    sens.modbus_data_updated()  # publish state to GUI and automations faster - assuming enabled, otherwise exception
-                except Exception:
-                    _LOGGER.debug(f"{self._name}: cannot send update for {key} - probably disabled ")
+            local_callback_needed = local_callback_needed or self.localsLoaded
 
-        if group.readFollowUp is not None:
-            if not await group.readFollowUp(self.data, data):
-                _LOGGER.warning("device group check not success")
-                return True
+        # Local controls can change independently while a Modbus group is being read.
+        for key in self.writeLocals:
+            if key in self.data:
+                data[key] = self.data[key]
 
-        # for key, value in data.items(): # remove for issue #1440, but then does not recognize communication errors anymore
-        #    self.data[key] = value # remove for issue #1440, but then comm errors are not detected
+        if res:
+            for key, descr in list(self.computedSensors.items()):
+                try:
+                    data[key] = descr.value_function(0, descr, data)
+                except Exception as ex:
+                    _LOGGER.debug(f"{self._name}: cannot compute value for {key}: {ex}")
+
+            if group.readFollowUp is not None:
+                if not await group.readFollowUp(previous_data, data):
+                    _LOGGER.warning(f"{self._name}: device group validation failed; discarding polling snapshot")
+                    return True
+
+            self._commit_poll_snapshot(previous_data, data)
+            if local_callback_needed:
+                self.plugin.localDataCallback(self)
+
+            for key, descr in list(self.computedSensors.items()):
+                sens = self.sensorEntities.get(key)
+                _LOGGER.debug(f"{self._name}: quickly updating state for computed sensor {sens} {key} {self.data.get(descr.key)} ")
+                if sens and (not descr.internal):
+                    try:
+                        sens.modbus_data_updated()
+                    except Exception:
+                        _LOGGER.debug(f"{self._name}: cannot send update for {key} - probably disabled ")
+            group.publish_updates = True
 
         if res and self.writequeue and self.plugin.isAwake(self.data):  # self.awakeplugin(self.data):
             # process outstanding write requests

tests/unit/test_poll_snapshot.py

@@ -0,0 +1,228 @@
+"""Tests for atomic polling snapshots."""
+
+import asyncio
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+from custom_components.solax_modbus import SolaXModbusHub
+
+
+def make_hub() -> Any:
+    """Build the minimal hub state required by polling tests."""
+    hub = cast(Any, object.__new__(SolaXModbusHub))
+    hub._name = "test"
+    hub.data = {"_repeatUntil": {}, "raw": 1}
+    hub.computedSensors = {}
+    hub.computedEntities = {}
+    hub.sensorEntities = {}
+    hub.writeLocals = {}
+    hub.writequeue = {}
+    hub.localsUpdated = False
+    hub.localsLoaded = True
+    hub.plugin = SimpleNamespace(
+        isAwake=Mock(return_value=True),
+        localDataCallback=Mock(return_value=True),
+    )
+    hub._poll_data_lock = asyncio.Lock()
+    hub.slowdown = 1
+    return hub
+
+
+def make_group(*, follow_up: Any = None) -> Any:
+    """Build a polling group with two holding-register blocks."""
+    return SimpleNamespace(
+        holdingBlocks=[
+            SimpleNamespace(start=1),
+            SimpleNamespace(start=2),
+        ],
+        inputBlocks=[],
+        readPreparation=None,
+        readFollowUp=follow_up,
+        publish_updates=False,
+        sensors=[],
+    )
+
+
+@pytest.mark.asyncio
+async def test_failed_group_discards_all_partial_values() -> None:
+    hub = make_hub()
+    group = make_group()
+    computed_sensor = Mock()
+    hub.computedSensors["computed"] = SimpleNamespace(
+        key="computed",
+        internal=False,
+        value_function=lambda initval, descr, data: data["raw"] * 2,
+    )
+    hub.sensorEntities["computed"] = computed_sensor
+
+    async def read_block(data: dict[str, Any], block: Any, typ: str) -> bool:
+        data["raw"] = block.start * 10
+        return bool(block.start == 1)
+
+    hub.async_read_modbus_block = read_block
+    original_data_object = hub.data
+
+    result = await hub.async_read_modbus_registers_all(group)
+
+    assert result is False
+    assert hub.data is original_data_object
+    assert hub.data["raw"] == 1
+    assert "computed" not in hub.data
+    assert group.publish_updates is False
+    computed_sensor.modbus_data_updated.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_tolerated_block_failure_commits_rest_of_snapshot() -> None:
+    hub = make_hub()
+    hub.data["unavailable"] = 5
+    group = make_group()
+
+    async def read_block(data: dict[str, Any], block: Any, typ: str) -> bool:
+        if block.start == 1:
+            data["raw"] = 10
+        else:
+            data.pop("unavailable", None)
+        return True
+
+    hub.async_read_modbus_block = read_block
+
+    result = await hub.async_read_modbus_registers_all(group)
+
+    assert result is True
+    assert hub.data["raw"] == 10
+    assert "unavailable" not in hub.data
+    assert group.publish_updates is True
+
+
+@pytest.mark.asyncio
+async def test_successful_group_commits_raw_and_computed_values_together() -> None:
+    hub = make_hub()
+    computed_sensor = Mock()
+    follow_up_observations: list[tuple[int, int, int]] = []
+
+    async def follow_up(old_data: dict[str, Any], new_data: dict[str, Any]) -> bool:
+        follow_up_observations.append((old_data["raw"], new_data["raw"], hub.data["raw"]))
+        return True
+
+    group = make_group(follow_up=follow_up)
+    hub.computedSensors["computed"] = SimpleNamespace(
+        key="computed",
+        internal=False,
+        value_function=lambda initval, descr, data: data["raw"] * 2,
+    )
+    hub.sensorEntities["computed"] = computed_sensor
+
+    async def read_block(data: dict[str, Any], block: Any, typ: str) -> bool:
+        data["raw"] = block.start
+        return True
+
+    hub.async_read_modbus_block = read_block
+    original_data_object = hub.data
+
+    result = await hub.async_read_modbus_registers_all(group)
+
+    assert result is True
+    assert follow_up_observations == [(1, 2, 1)]
+    assert hub.data is original_data_object
+    assert hub.data["raw"] == 2
+    assert hub.data["computed"] == 4
+    assert group.publish_updates is True
+    computed_sensor.modbus_data_updated.assert_called_once_with()
+
+
+@pytest.mark.asyncio
+async def test_failed_follow_up_discards_snapshot_without_publishing() -> None:
+    hub = make_hub()
+    computed_sensor = Mock()
+    group = make_group(follow_up=AsyncMock(return_value=False))
+    hub.computedSensors["computed"] = SimpleNamespace(
+        key="computed",
+        internal=False,
+        value_function=lambda initval, descr, data: data["raw"] * 2,
+    )
+    hub.sensorEntities["computed"] = computed_sensor
+
+    async def read_block(data: dict[str, Any], block: Any, typ: str) -> bool:
+        data["raw"] = 9
+        return True
+
+    hub.async_read_modbus_block = read_block
+
+    result = await hub.async_read_modbus_registers_all(group)
+
+    assert result is True
+    assert hub.data["raw"] == 1
+    assert "computed" not in hub.data
+    assert group.publish_updates is False
+    computed_sensor.modbus_data_updated.assert_not_called()
+
+
+def test_snapshot_commit_preserves_concurrent_local_change() -> None:
+    hub = make_hub()
+    previous_data = {"_repeatUntil": {}, "raw": 1, "removed": 5}
+    new_data = {"_repeatUntil": {}, "raw": 2, "added": 7}
+    hub.data = {"_repeatUntil": {}, "raw": 99, "removed": 5}
+
+    hub._commit_poll_snapshot(previous_data, new_data)
+
+    assert hub.data["raw"] == 99
+    assert hub.data["added"] == 7
+    assert "removed" not in hub.data
+
+
+@pytest.mark.asyncio
+async def test_group_reads_are_serialized() -> None:
+    hub = make_hub()
+    active_reads = 0
+    maximum_active_reads = 0
+
+    async def read_group(group: Any) -> bool:
+        nonlocal active_reads, maximum_active_reads
+        active_reads += 1
+        maximum_active_reads = max(maximum_active_reads, active_reads)
+        await asyncio.sleep(0)
+        active_reads -= 1
+        group.publish_updates = True
+        return True
+
+    hub.async_read_modbus_registers_all = read_group
+    first_group = make_group()
+    second_group = make_group()
+
+    first_result, second_result = await asyncio.gather(
+        hub.async_read_modbus_data(first_group),
+        hub.async_read_modbus_data(second_group),
+    )
+
+    assert first_result is True
+    assert second_result is True
+    assert maximum_active_reads == 1
+
+
+@pytest.mark.asyncio
+async def test_successful_but_discarded_snapshot_does_not_publish_group() -> None:
+    hub = make_hub()
+    sensor = Mock()
+    group = make_group()
+    group.sensors = [sensor]
+    interval_group = SimpleNamespace(device_groups={"test": group})
+    hub.blocks_changed = False
+    hub.cyclecount = 1
+    hub.sleepnone = []
+    hub.sleepzero = []
+
+    async def read_group(current_group: Any) -> bool:
+        current_group.publish_updates = False
+        return True
+
+    hub.async_read_modbus_data = read_group
+
+    result, updated_sensors = await hub._refresh_interval_group_once(interval_group)
+
+    assert result is True
+    assert updated_sensors == 0
+    sensor.modbus_data_updated.assert_not_called()