Use PriorityQueue.qsize() to obtain the queue depth

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()