Implement priority queue for sending messages

Author: brefra

plugwise/constants.py

@@ -136,6 +136,11 @@
 # Default sleep between sending messages
 SLEEP_TIME = 150 / 1000
 
+# Message priority levels
+PRIORITY_HIGH = 1
+PRIORITY_LOW = 2
+PRIORITY_MEDIUM = 3
+
 # Max seconds the internal clock of plugwise nodes
 # are allowed to drift in seconds
 MAX_TIME_DRIFT = 30

plugwise/controller.py

@@ -12,8 +12,8 @@
 """
 
 from datetime import datetime, timedelta
+from queue import Empty, PriorityQueue
 import logging
-import queue
 import threading
 import time
 
@@ -22,6 +22,7 @@
 from .constants import (
     MESSAGE_RETRY,
     MESSAGE_TIME_OUT,
+    PRIORITY_MEDIUM,
     REQUEST_FAILED,
     REQUEST_SUCCESS,
     SLEEP_TIME,
@@ -77,7 +78,7 @@ def connect_to_stick(self, callback=None) -> bool:
         if self.connection.connect():
             _LOGGER.debug("Starting message controller threads...")
             # send daemon
-            self._send_message_queue = queue.Queue()
+            self._send_message_queue = PriorityQueue()
             self._run_send_message_thread = True
             self._send_message_thread = threading.Thread(
                 None, self._send_message_loop, "send_messages_thread", (), {}
@@ -96,20 +97,32 @@ def connect_to_stick(self, callback=None) -> bool:
             _LOGGER.warning("Failed to connect to USB stick")
         return self.connection.is_connected()
 
-    def send(self, request: NodeRequest, callback=None, retry_counter=0):
+    def send(
+        self,
+        request: NodeRequest,
+        callback=None,
+        retry_counter=0,
+        priority=PRIORITY_MEDIUM,
+    ):
         """Queue request message to be sent into Plugwise Zigbee network."""
         _LOGGER.debug(
-            "Queue %s to be send with retry counter %s",
+            "Queue %s to be send with retry counter %s and priority %s",
             request.__class__.__name__,
             str(retry_counter),
+            str(priority),
         )
         self._send_message_queue.put(
-            [
-                request,
-                callback,
+            (
+                priority,
                 retry_counter,
-                None,
-            ]
+                datetime.now(),
+                [
+                    request,
+                    callback,
+                    retry_counter,
+                    None,
+                ],
+            )
         )
 
     def resend(self, seq_id):
@@ -180,8 +193,10 @@ def _send_message_loop(self):
         """Daemon to send messages waiting in queue."""
         while self._run_send_message_thread:
             try:
-                request_set = self._send_message_queue.get(block=True, timeout=1)
-            except queue.Empty:
+                _prio, _retry, _dt, request_set = self._send_message_queue.get(
+                    block=True, timeout=1
+                )
+            except Empty:
                 time.sleep(SLEEP_TIME)
             else:
                 # Calc next seq_id based last received ack message

plugwise/nodes/__init__.py

@@ -8,6 +8,7 @@
     FEATURE_RELAY,
     FEATURE_RSSI_IN,
     FEATURE_RSSI_OUT,
+    PRIORITY_LOW,
     UTF8_DECODE,
 )
 from ..messages.requests import NodeFeaturesRequest, NodeInfoRequest, NodePingRequest
@@ -155,6 +156,8 @@ def _request_info(self, callback=None):
         self.message_sender(
             NodeInfoRequest(self._mac),
             callback,
+            0,
+            PRIORITY_LOW,
         )
 
     def _request_features(self, callback=None):

plugwise/nodes/circle.py

@@ -17,6 +17,8 @@
     FEATURE_RSSI_OUT,
     MAX_TIME_DRIFT,
     MESSAGE_TIME_OUT,
+    PRIORITY_HIGH,
+    PRIORITY_LOW,
     PULSES_PER_KW_SECOND,
     RELAY_SWITCHED_OFF,
     RELAY_SWITCHED_ON,
@@ -163,13 +165,17 @@ def _request_calibration(self, callback=None):
         self.message_sender(
             CircleCalibrationRequest(self._mac),
             callback,
+            0,
+            PRIORITY_HIGH,
         )
 
     def _request_switch(self, state, callback=None):
         """Request to switch relay state and request state info"""
         self.message_sender(
             CircleSwitchRelayRequest(self._mac, state),
             callback,
+            0,
+            PRIORITY_HIGH,
         )
 
     def _request_power_update(self, callback=None):
@@ -358,21 +364,31 @@ def request_power_buffer(self, log_address=None, callback=None):
                 # Only request last 2 power buffer logs
                 self.message_sender(
                     CirclePowerBufferRequest(self._mac, log_address - 1),
+                    None,
+                    0,
+                    PRIORITY_LOW,
                 )
                 self.message_sender(
                     CirclePowerBufferRequest(self._mac, log_address),
                     callback,
+                    0,
+                    PRIORITY_LOW,
                 )
             else:
                 # Collect power history info of today and yesterday
                 # Each request contains 4 hours except last request
                 for req_log_address in range(log_address - 13, log_address):
                     self.message_sender(
                         CirclePowerBufferRequest(self._mac, req_log_address),
+                        None,
+                        0,
+                        PRIORITY_LOW,
                     )
                 self.message_sender(
                     CirclePowerBufferRequest(self._mac, log_address),
                     callback,
+                    0,
+                    PRIORITY_LOW,
                 )
 
     def _response_power_buffer(self, message):
@@ -452,6 +468,8 @@ def get_clock(self, callback=None):
         self.message_sender(
             CircleClockGetRequest(self._mac),
             callback,
+            0,
+            PRIORITY_LOW,
         )
 
     def set_clock(self, callback=None):

plugwise/nodes/circle_plus.py

@@ -2,7 +2,7 @@
 from datetime import datetime
 import logging
 
-from ..constants import MAX_TIME_DRIFT, UTF8_DECODE
+from ..constants import MAX_TIME_DRIFT, PRIORITY_LOW, UTF8_DECODE
 from ..messages.requests import (
     CirclePlusRealTimeClockGetRequest,
     CirclePlusRealTimeClockSetRequest,
@@ -92,6 +92,8 @@ def get_real_time_clock(self, callback=None):
         self.message_sender(
             CirclePlusRealTimeClockGetRequest(self._mac),
             callback,
+            0,
+            PRIORITY_LOW,
         )
 
     def _response_realtime_clock(self, message):

plugwise/nodes/sed.py

@@ -10,6 +10,7 @@
     FEATURE_PING,
     FEATURE_RSSI_IN,
     FEATURE_RSSI_OUT,
+    PRIORITY_HIGH,
     SED_AWAKE_BUTTON,
     SED_AWAKE_FIRST,
     SED_AWAKE_MAINTENANCE,
@@ -90,7 +91,7 @@ def _process_awake_response(self, message):
                     request_message.__class__.__name__,
                     self.mac,
                 )
-                self.message_sender(request_message, callback, -1)
+                self.message_sender(request_message, callback, -1, PRIORITY_HIGH)
             self._SED_requests = {}
         else:
             if message.awake_type.value == SED_AWAKE_STATE:

plugwise/stick.py

@@ -23,6 +23,7 @@
     NODE_TYPE_SENSE,
     NODE_TYPE_STEALTH,
     NODE_TYPE_SWITCH,
+    PRIORITY_LOW,
     STATE_ACTIONS,
     UTF8_DECODE,
     WATCHDOG_DEAMON,
@@ -656,6 +657,7 @@ def _update_loop(self):
                             NodePingRequest(bytes(mac, UTF8_DECODE)),
                             None,
                             -1,
+                            PRIORITY_LOW,
                         )
                     _discover_counter = 0
                 else:
@@ -679,6 +681,7 @@ def auto_update(self, timer=None):
         """Configure auto update polling daemon for power usage and availability state."""
         if timer:
             self._auto_update_timer = timer
+            self._auto_update_manually = True
         elif timer == 0:
             self._auto_update_timer = 0
             self._run_update_thread = False
@@ -764,6 +767,8 @@ def discover_node(self, mac: str, callback=None, force_discover=False) -> bool:
                         self.msg_controller.send(
                             NodeInfoRequest(bytes(mac, UTF8_DECODE)),
                             callback,
+                            0,
+                            PRIORITY_LOW,
                         )
                     elif force_discover:
                         self.msg_controller.send(