Merge pull request #291 from plugwise/alt_prio_queue

Use PriorityQueue.qsize() to obtain the queue depth

Author: bouwew

CHANGELOG.md

@@ -1,7 +1,8 @@
 # Changelog
 
-## Ongoing / v0.44.8a0
+## v0.44.8
 
+- PR [291](https://github.com/plugwise/python-plugwise-usb/pull/291): Collect send-queue depth via PriorityQueue.qsize(), this provides a more accurate result
 - Fix for [#288](https://github.com/plugwise/plugwise_usb-beta/issues/288) via PR [293](https://github.com/plugwise/python-plugwise-usb/pull/293)
 - Chores move module publishing on (test)pypi to Trusted Publishing (and using uv) - released as alpha 0.44.8a0 to demonstrate functionality
 

plugwise_usb/connection/manager.py

@@ -36,15 +36,6 @@ def __init__(self) -> None:
         ] = {}
         self._unsubscribe_stick_events: Callable[[], None] | None = None
 
-    @property
-    def queue_depth(self) -> int:
-        """Return estimated size of pending responses."""
-        return self._sender.processed_messages - self._receiver.processed_messages
-
-    def correct_received_messages(self, correction: int) -> None:
-        """Correct received messages count."""
-        self._receiver.correct_processed_messages(correction)
-
     @property
     def serial_path(self) -> str:
         """Return current port."""

plugwise_usb/connection/queue.py

@@ -8,6 +8,7 @@
 import logging
 
 from ..api import StickEvent
+from ..constants import REPORT_QUEUE_FILLING_UP
 from ..exceptions import MessageError, NodeTimeout, StickError, StickTimeout
 from ..messages import Priority
 from ..messages.requests import NodePingRequest, PlugwiseCancelRequest, PlugwiseRequest
@@ -112,13 +113,11 @@ async def submit(self, request: PlugwiseRequest) -> PlugwiseResponse | None:
                     _LOGGER.warning("%s, cancel request", exc)  # type: ignore[unreachable]
             except StickError as exc:
                 _LOGGER.error(exc)
-                self._stick.correct_received_messages(1)
                 raise StickError(
                     f"No response received for {request.__class__.__name__} "
                     + f"to {request.mac_decoded}"
                 ) from exc
             except BaseException as exc:
-                self._stick.correct_received_messages(1)
                 raise StickError(
                     f"No response received for {request.__class__.__name__} "
                     + f"to {request.mac_decoded}"
@@ -145,12 +144,12 @@ async def _send_queue_worker(self) -> None:
                 self._submit_queue.task_done()
                 return
 
-            if self._stick.queue_depth > 3:
+            qsize = self._submit_queue.qsize()
+            if qsize > REPORT_QUEUE_FILLING_UP:
+                # When the queue size grows, rate-limit the sending of requests to avoid overloading the network
                 await sleep(0.125)
-                if self._stick.queue_depth > 3:
-                    _LOGGER.warning(
-                        "Awaiting plugwise responses %d", self._stick.queue_depth
-                    )
+                if qsize > REPORT_QUEUE_FILLING_UP:
+                    _LOGGER.warning("Awaiting plugwise responses %d", qsize)
 
             await self._stick.write_to_stick(request)
             self._submit_queue.task_done()

plugwise_usb/connection/receiver.py

@@ -99,7 +99,6 @@ 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
@@ -138,20 +137,11 @@ 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:
-        """Correct 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")
@@ -291,7 +281,6 @@ 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")

plugwise_usb/connection/sender.py

@@ -38,17 +38,11 @@ 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.
@@ -149,7 +143,6 @@ 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/constants.py

@@ -19,6 +19,7 @@
 # Value limits
 MAX_UINT_2: Final = 255  # 8-bit unsigned integer max
 MAX_UINT_4: Final = 65535  # 16-bit unsigned integer max
+REPORT_QUEUE_FILLING_UP: Final = 8
 
 # Time
 DAY_IN_HOURS: Final = 24

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name            = "plugwise_usb"
-version         = "0.44.7"
+version         = "0.44.8a8"
 license         = "MIT"
 keywords        = ["home", "automation", "plugwise", "module", "usb"]
 classifiers     = [

tests/test_usb.py

@@ -1691,7 +1691,6 @@ async def makedirs(cache_dir: str, exist_ok: bool) -> None:
                 "FEDCBA9876543210": pw_api.NodeType.CIRCLE,
                 "1298347650AFBECD": pw_api.NodeType.SCAN,
             }
-        pw_nw_cache.update_nodetypes("1234ABCD4321FEDC", pw_api.NodeType.STEALTH)
 
         with patch("aiofiles.threadpool.sync_open", return_value=mock_file_stream):
             # await pw_nw_cache.save_cache()