keep repeating discover_nodes until all registered nodes are online

.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 @@
  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 @@
  # endregion
 
  # region - Nodes
- def _create_node_object(
+async def _create_node_object(
  self,
  mac: str,
  address: int,
@@ -363,7 +371,7 @@
  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 @@
 
  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_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 @@
  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 @@
  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 @@
  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 @@
  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)