Correct message overload on intitial circle log fetch.

Limit outstanding messages to 4

Author: ArnoutD

plugwise_usb/connection/manager.py

@@ -36,6 +36,13 @@ def __init__(self) -> None:
         ] = {}
         self._unsubscribe_stick_events: Callable[[], None] | None = None
 
+    @property
+    def queue_depth(self) -> int:
+        return self._sender.processed_messages - self._receiver.processed_messages
+
+    def correct_received_messages(self, correction: int) -> None:
+        self._receiver.correct_processed_messages(correction)
+
     @property
     def serial_path(self) -> str:
         """Return current port."""

plugwise_usb/connection/queue.py

@@ -82,7 +82,7 @@ async def submit(self, request: PlugwiseRequest) -> PlugwiseResponse:
             )
 
         while request.resend and not request.waiting_for_response:
-            _LOGGER.warning("submit | start (%s) %s", request.retries_left, request)
+            _LOGGER.debug("submit | start (%s) %s", request.retries_left, request)
             if not self._running or self._stick is None:
                 raise StickError(
                     f"Cannot send message {request.__class__.__name__} for"
@@ -91,6 +91,7 @@ async def submit(self, request: PlugwiseRequest) -> PlugwiseResponse:
             await self._add_request_to_queue(request)
             try:
                 response: PlugwiseResponse = await request.response_future()
+                return response
             except (NodeTimeout, StickTimeout) as e:
                 if isinstance(request, NodePingRequest):
                     # For ping requests it is expected to receive timeouts, so lower log level
@@ -103,17 +104,19 @@ async def submit(self, request: PlugwiseRequest) -> PlugwiseResponse:
                     _LOGGER.warning("%s, cancel request", e)  # type: ignore[unreachable]
             except StickError as exception:
                 _LOGGER.error(exception)
+                self._stick.correct_received_messages(1)
                 raise StickError(
                     f"No response received for {request.__class__.__name__} "
                     + f"to {request.mac_decoded}"
                 ) from exception
             except BaseException as exception:
+                self._stick.correct_received_messages(1)
                 raise StickError(
                     f"No response received for {request.__class__.__name__} "
                     + f"to {request.mac_decoded}"
                 ) from exception
 
-        return response
+        return None
 
     async def _add_request_to_queue(self, request: PlugwiseRequest) -> None:
         """Add request to send queue."""
@@ -133,8 +136,13 @@ async def _send_queue_worker(self) -> None:
             if request.priority == Priority.CANCEL:
                 self._submit_queue.task_done()
                 return
+
+            while self._stick.queue_depth > 3:
+                _LOGGER.info("Awaiting plugwise responses %d", self._stick.queue_depth)
+                await sleep(0.125)
+
             await self._stick.write_to_stick(request)
             self._submit_queue.task_done()
-            await sleep(0.001)
+
             _LOGGER.debug("Sent from queue %s", request)
         _LOGGER.debug("Send_queue_worker stopped")

plugwise_usb/connection/receiver.py

@@ -99,6 +99,7 @@ def __init__(
         self._data_worker_task: Task[None] | None = None
 
         # Message processing
+        self._processed_msgs = 0
         self._message_queue: PriorityQueue[PlugwiseResponse] = PriorityQueue()
         self._last_processed_messages: list[bytes] = []
         self._current_seq_id: bytes | None = None
@@ -137,11 +138,20 @@ def connection_lost(self, exc: Exception | None = None) -> None:
         self._transport = None
         self._connection_state = False
 
+    @property
+    def processed_messages(self) -> int:
+        """Return the number of processed messages."""
+        return self._processed_msgs
+
     @property
     def is_connected(self) -> bool:
         """Return current connection state of the USB-Stick."""
         return self._connection_state
 
+    def correct_processed_messages(self, correction: int) -> None:
+        """Return the number of processed messages."""
+        self._processed_msgs += correction
+
     def connection_made(self, transport: SerialTransport) -> None:
         """Call when the serial connection to USB-Stick is established."""
         _LOGGER.info("Connection made")
@@ -278,6 +288,7 @@ async def _message_queue_worker(self) -> None:
                 await self._notify_stick_subscribers(response)
             else:
                 await self._notify_node_response_subscribers(response)
+                self._processed_msgs += 1
             self._message_queue.task_done()
             await sleep(0)
         _LOGGER.debug("Message queue worker stopped")
@@ -457,7 +468,7 @@ async def _notify_node_response_subscribers(
 
         self._node_subscription_lock.release()
         if len(notify_tasks) > 0:
-            _LOGGER.debug("Received %s", node_response)
+            _LOGGER.debug("Received %s %s", node_response, node_response.seq_id)
             if node_response.seq_id not in BROADCAST_IDS:
                 self._last_processed_messages.append(node_response.seq_id)
             # Limit tracking to only the last appended request (FIFO)

plugwise_usb/connection/sender.py

@@ -38,11 +38,17 @@ def __init__(self, stick_receiver: StickReceiver, transport: Transport) -> None:
         self._loop = get_running_loop()
         self._receiver = stick_receiver
         self._transport = transport
+        self._processed_msgs = 0
         self._stick_response: Future[StickResponse] | None = None
         self._stick_lock = Lock()
         self._current_request: None | PlugwiseRequest = None
         self._unsubscribe_stick_response: Callable[[], None] | None = None
 
+    @property
+    def processed_messages(self) -> int:
+        """Return the number of processed messages."""
+        return self._processed_msgs
+    
     async def start(self) -> None:
         """Start the sender."""
         # Subscribe to ACCEPT stick responses, which contain the seq_id we need.
@@ -133,6 +139,7 @@ async def write_request_to_port(self, request: PlugwiseRequest) -> None:
         finally:
             self._stick_response.cancel()
             self._stick_lock.release()
+            self._processed_msgs += 1
 
     async def _process_stick_response(self, response: StickResponse) -> None:
         """Process stick response."""

plugwise_usb/nodes/circle.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from asyncio import Task, create_task, gather
+from asyncio import Task, create_task
 from collections.abc import Awaitable, Callable
 from dataclasses import replace
 from datetime import UTC, datetime
@@ -453,11 +453,8 @@ async def get_missing_energy_logs(self) -> None:
                 log_address, _ = calc_log_address(log_address, 1, -4)
                 total_addresses -= 1
 
-            if not all(await gather(*log_update_tasks)):
-                _LOGGER.info(
-                    "Failed to request one or more update energy log for %s",
-                    self._mac_in_str,
-                )
+            for task in log_update_tasks:
+                await task
 
             if self._cache_enabled:
                 await self._energy_log_records_save_to_cache()
@@ -475,9 +472,8 @@ async def get_missing_energy_logs(self) -> None:
             )
 
             missing_addresses = sorted(missing_addresses, reverse=True)
-            await gather(
-                *[self.energy_log_update(address) for address in missing_addresses]
-            )
+            for address in missing_addresses:
+                await self.energy_log_update(address)
 
         if self._cache_enabled:
             await self._energy_log_records_save_to_cache()
@@ -528,7 +524,7 @@ async def _energy_log_records_load_from_cache(self) -> bool:
         """Load energy_log_record from cache."""
         cache_data = self._get_cache(CACHE_ENERGY_COLLECTION)
         if (cache_data := self._get_cache(CACHE_ENERGY_COLLECTION)) is None:
-            _LOGGER.debug(
+            _LOGGER.warning(
                 "Failed to restore energy log records from cache for node %s", self.name
             )
             return False
@@ -811,7 +807,7 @@ async def _load_from_cache(self) -> bool:
             return False
         # Energy collection
         if await self._energy_log_records_load_from_cache():
-            _LOGGER.debug(
+            _LOGGER.warning(
                 "Node %s failed to load energy_log_records from cache",
                 self._mac_in_str,
             )