Merge pull request #279 from plugwise/mdi

keep repeating discover_nodes until all registered nodes are online

Author: dirixmjm

.gitignore

@@ -19,3 +19,4 @@ fixtures/*
 tmp
 .cache
 appdata_folder
+mock_folder_that_exists/*

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## v0.44.6 - 2025-07-06
+
+- PR [279](https://github.com/plugwise/python-plugwise-usb/pull/279): Improve registry cache and node load behaviour
+
 ## v0.44.5 - 2025-06-22
 
 - PR [274](https://github.com/plugwise/python-plugwise-usb/pull/274): Make the energy-reset function available to Plus devices

plugwise_usb/constants.py

@@ -40,6 +40,10 @@
 # In bigger networks a response from a Node could take up a while, so lets use 15 seconds.
 NODE_TIME_OUT: Final = 15
 
+# Retry delay discover nodes
+NODE_RETRY_DISCOVER_INTERVAL = 60
+NODE_RETRY_LOAD_INTERVAL = 60
+
 MAX_RETRIES: Final = 3
 SUPPRESS_INITIALIZATION_WARNINGS: Final = 10  # Minutes to suppress (expected) communication warning messages after initialization
 

plugwise_usb/network/__init__.py

@@ -12,7 +12,12 @@
 
 from ..api import NodeEvent, NodeType, PlugwiseNode, StickEvent
 from ..connection import StickController
-from ..constants import ENERGY_NODE_TYPES, UTF8
+from ..constants import (
+    ENERGY_NODE_TYPES,
+    NODE_RETRY_DISCOVER_INTERVAL,
+    NODE_RETRY_LOAD_INTERVAL,
+    UTF8,
+)
 from ..exceptions import CacheError, MessageError, NodeError, StickError, StickTimeout
 from ..helpers.util import validate_mac
 from ..messages.requests import CircleMeasureIntervalRequest, NodePingRequest
@@ -72,6 +77,9 @@ def __init__(
         self._unsubscribe_node_rejoin: Callable[[], None] | None = None
 
         self._discover_sed_tasks: dict[str, Task[bool]] = {}
+        self._registry_stragglers: dict[int, str] = {}
+        self._discover_stragglers_task: Task[None] | None = None
+        self._load_stragglers_task: Task[None] | None = None
 
     # region - Properties
 
@@ -338,7 +346,7 @@ async def discover_network_coordinator(self, load: bool = False) -> bool:
     # endregion
 
     # region - Nodes
-    def _create_node_object(
+    async def _create_node_object(
         self,
         mac: str,
         address: int,
@@ -363,7 +371,7 @@ def _create_node_object(
             return
         self._nodes[mac] = node
         _LOGGER.debug("%s node %s added", node.__class__.__name__, mac)
-        self._register.update_network_registration(address, mac, node_type)
+        await self._register.update_network_registration(address, mac, node_type)
 
         if self._cache_enabled:
             _LOGGER.debug(
@@ -404,22 +412,24 @@ async def _discover_node(
 
         Return True if discovery succeeded.
         """
-        _LOGGER.debug("Start discovery of node %s ", mac)
+        _LOGGER.debug(
+            "Start discovery of node %s with NodeType %s", mac, str(node_type)
+        )
         if self._nodes.get(mac) is not None:
             _LOGGER.debug("Skip discovery of already known node %s ", mac)
             return True
 
         if node_type is not None:
-            self._create_node_object(mac, address, node_type)
+            await self._create_node_object(mac, address, node_type)
             await self._notify_node_event_subscribers(NodeEvent.DISCOVERED, mac)
             return True
 
         # Node type is unknown, so we need to discover it first
-        _LOGGER.debug("Starting the discovery of node %s", mac)
+        _LOGGER.debug("Starting the discovery of node %s with unknown NodeType", mac)
         node_info, node_ping = await self._controller.get_node_details(mac, ping_first)
         if node_info is None:
             return False
-        self._create_node_object(mac, address, node_info.node_type)
+        await self._create_node_object(mac, address, node_info.node_type)
 
         # Forward received NodeInfoResponse message to node
         await self._nodes[mac].message_for_node(node_info)
@@ -431,15 +441,39 @@ async def _discover_node(
     async def _discover_registered_nodes(self) -> None:
         """Discover nodes."""
         _LOGGER.debug("Start discovery of registered nodes")
-        counter = 0
+        registered_counter = 0
         for address, registration in self._register.registry.items():
             mac, node_type = registration
             if mac != "":
                 if self._nodes.get(mac) is None:
-                    await self._discover_node(address, mac, node_type)
-                counter += 1
+                    if not await self._discover_node(address, mac, node_type):
+                        self._registry_stragglers[address] = mac
+                registered_counter += 1
                 await sleep(0)
-        _LOGGER.debug("Total %s registered node(s)", str(counter))
+        if len(self._registry_stragglers) > 0 and (
+            self._discover_stragglers_task is None
+            or self._discover_stragglers_task.done()
+        ):
+            self._discover_stragglers_task = create_task(self._discover_stragglers())
+        _LOGGER.debug(
+            "Total %s online of %s registered node(s)",
+            str(len(self._nodes)),
+            str(registered_counter),
+        )
+
+    async def _discover_stragglers(self) -> None:
+        """Repeat Discovery of Nodes with unknown NodeType."""
+        while len(self._registry_stragglers) > 0:
+            await sleep(NODE_RETRY_DISCOVER_INTERVAL)
+            stragglers: dict[int, str] = {}
+            for address, mac in self._registry_stragglers.items():
+                if not await self._discover_node(address, mac, None):
+                    stragglers[address] = mac
+            self._registry_stragglers = stragglers
+            _LOGGER.debug(
+                "Total %s nodes unreachable having unknown NodeType",
+                str(len(stragglers)),
+            )
 
     async def _load_node(self, mac: str) -> bool:
         """Load node."""
@@ -452,6 +486,12 @@ async def _load_node(self, mac: str) -> bool:
             return True
         return False
 
+    async def _load_stragglers(self) -> None:
+        """Retry failed load operation."""
+        await sleep(NODE_RETRY_LOAD_INTERVAL)
+        while not self._load_discovered_nodes():
+            await sleep(NODE_RETRY_LOAD_INTERVAL)
+
     async def _load_discovered_nodes(self) -> bool:
         """Load all nodes currently discovered."""
         _LOGGER.debug("_load_discovered_nodes | START | %s", len(self._nodes))
@@ -499,10 +539,10 @@ async def discover_nodes(self, load: bool = True) -> bool:
         await self.discover_network_coordinator(load=load)
         if not self._is_running:
             await self.start()
-
         await self._discover_registered_nodes()
-        if load:
-            return await self._load_discovered_nodes()
+        if load and not await self._load_discovered_nodes():
+            self._load_stragglers_task = create_task(self._load_stragglers())
+            return False
 
         return True
 
@@ -512,10 +552,22 @@ async def stop(self) -> None:
         for task in self._discover_sed_tasks.values():
             if not task.done():
                 task.cancel()
+        if (
+            hasattr(self, "_load_stragglers_task")
+            and self._load_stragglers_task
+            and not self._load_stragglers_task.done()
+        ):
+            self._load_stragglers_task.cancel()
+        if (
+            hasattr(self, "_discover_stragglers_task")
+            and self._discover_stragglers_task
+            and not self._discover_stragglers_task.done()
+        ):
+            self._discover_stragglers_task.cancel()
         self._is_running = False
         self._unsubscribe_to_protocol_events()
         await self._unload_discovered_nodes()
-        await self._register.stop()
+        self._register.stop()
         _LOGGER.debug("Stopping finished")
 
     # endregion

plugwise_usb/network/cache.py

@@ -5,11 +5,10 @@
 import logging
 
 from ..api import NodeType
-from ..constants import CACHE_DATA_SEPARATOR
 from ..helpers.cache import PlugwiseCache
 
 _LOGGER = logging.getLogger(__name__)
-_NETWORK_CACHE_FILE_NAME = "nodes.cache"
+_NETWORK_CACHE_FILE_NAME = "nodetype.cache"
 
 
 class NetworkRegistrationCache(PlugwiseCache):
@@ -18,68 +17,63 @@ class NetworkRegistrationCache(PlugwiseCache):
     def __init__(self, cache_root_dir: str = "") -> None:
         """Initialize NetworkCache class."""
         super().__init__(_NETWORK_CACHE_FILE_NAME, cache_root_dir)
-        self._registrations: dict[int, tuple[str, NodeType | None]] = {}
+        self._nodetypes: dict[str, NodeType] = {}
 
     @property
-    def registrations(self) -> dict[int, tuple[str, NodeType | None]]:
+    def nodetypes(self) -> dict[str, NodeType]:
         """Cached network information."""
-        return self._registrations
+        return self._nodetypes
 
     async def save_cache(self) -> None:
         """Save the node information to file."""
         cache_data_to_save: dict[str, str] = {}
-        for address in range(-1, 64, 1):
-            mac, node_type = self._registrations.get(address, ("", None))
-            if node_type is None:
-                node_value = ""
-            else:
-                node_value = str(node_type)
-            cache_data_to_save[str(address)] = (
-                f"{mac}{CACHE_DATA_SEPARATOR}{node_value}"
-            )
+        for mac, node_type in self._nodetypes.items():
+            node_value = str(node_type)
+            cache_data_to_save[mac] = node_value
+        _LOGGER.debug("Save NodeTypes %s", str(len(cache_data_to_save)))
         await self.write_cache(cache_data_to_save)
 
     async def clear_cache(self) -> None:
         """Clear current cache."""
-        self._registrations = {}
+        self._nodetypes = {}
         await self.delete_cache()
 
     async def restore_cache(self) -> None:
         """Load the previously stored information."""
         data: dict[str, str] = await self.read_cache()
-        self._registrations = {}
-        for _key, _data in data.items():
-            address = int(_key)
-            try:
-                if CACHE_DATA_SEPARATOR in _data:
-                    values = _data.split(CACHE_DATA_SEPARATOR)
-                else:
-                    # legacy data separator can by remove at next version
-                    values = _data.split(";")
-                mac = values[0]
-                node_type: NodeType | None = None
-                if values[1] != "":
-                    node_type = NodeType[values[1][9:]]
-                self._registrations[address] = (mac, node_type)
-                _LOGGER.debug(
-                    "Restore registry address %s with mac %s with node type %s",
-                    address,
-                    mac if mac != "" else "<empty>",
-                    str(node_type),
-                )
-            except (KeyError, IndexError):
-                _LOGGER.warning(
-                    "Skip invalid data '%s' in cache file '%s'",
-                    _data,
-                    self._cache_file,
-                )
+        self._nodetypes = {}
+        for mac, node_value in data.items():
+            node_type: NodeType | None = None
+            if len(node_value) >= 10:
+                try:
+                    node_type = NodeType[node_value[9:]]
+                except KeyError:
+                    node_type = None
+            if node_type is None:
+                _LOGGER.warning("Invalid NodeType in cache: %s", node_value)
+                continue
+            self._nodetypes[mac] = node_type
+            _LOGGER.debug(
+                "Restore NodeType for mac %s with node type %s",
+                mac,
+                str(node_type),
+            )
 
-    def update_registration(
-        self, address: int, mac: str, node_type: NodeType | None
-    ) -> None:
+    async def update_nodetypes(self, mac: str, node_type: NodeType | None) -> None:
         """Save node information in cache."""
-        if self._registrations.get(address) is not None:
-            _, current_node_type = self._registrations[address]
-            if current_node_type is not None and node_type is None:
+        if node_type is None:
+            return
+        if (current_node_type := self._nodetypes.get(mac)) is not None:
+            if current_node_type == node_type:
                 return
-        self._registrations[address] = (mac, node_type)
+            _LOGGER.warning(
+                "Cache contained mismatched NodeType %s replacing with %s",
+                str(current_node_type),
+                str(node_type),
+            )
+        self._nodetypes[mac] = node_type
+        await self.save_cache()
+
+    def get_nodetype(self, mac: str) -> NodeType | None:
+        """Return NodeType from cache."""
+        return self._nodetypes.get(mac)