detect some HID++1.0 custom battery notifications

they're not handled yet

lib/logitech/unifying_receiver/base.py

@@ -229,10 +229,18 @@ def make_notification(devnumber, data):
 	return a Notification tuple if it is."""
 	sub_id = ord(data[:1])
 	if sub_id & 0x80 != 0x80:
-		# HID++ 1.0 standard notifications are 0x40 - 0x7F
-		# HID++ 2.0 feature notifications have the SoftwareID 0
+		# if this is not a HID++1.0 register r/w
 		address = ord(data[1:2])
-		if sub_id >= 0x40 or address & 0x0F == 0x00:
+		if (
+			# standard HID++ 1.0 notification, SubId may be 0x40 - 0x7F
+			(sub_id >= 0x40)
+			or
+			# custom HID++1.0 battery events, where SubId is 0x07/0x0D
+			(sub_id in (0x07, 0x0D) and len(data) == 5 and data[4:5] == b'\x00')
+			or
+			# HID++ 2.0 feature notifications have the SoftwareID 0
+			(address & 0x0F == 0x00)
+			):
 			return _HIDPP_Notification(devnumber, sub_id, address, data[2:])
 
 from collections import namedtuple

lib/logitech/unifying_receiver/listener.py

@@ -59,8 +59,7 @@ class _ThreadedHandle(object):
 			handles, self._handles = self._handles, []
 			if _log.isEnabledFor(_DEBUG):
 				_log.debug("%s closing %s", repr(self), handles)
-			for h in handles:
-				_base.close(h)
+			map(_base.close, handles)
 
 	@property
 	def notifications_hook(self):
@@ -87,7 +86,7 @@ class _ThreadedHandle(object):
 	__unicode__ = __str__
 
 	def __repr__(self):
-		return '<ThreadedHandle(%s)>' % self.path
+		return '<_ThreadedHandle(%s)>' % self.path
 
 	def __bool__(self):
 		return bool(self._local)
@@ -104,7 +103,7 @@ class _ThreadedHandle(object):
 _EVENT_READ_TIMEOUT = 500
 
 # After this many reads that did not produce a packet, call the tick() method.
-_IDLE_READS = 4
+_IDLE_READS = 3
 
 
 class EventsListener(_threading.Thread):
@@ -163,9 +162,8 @@ class EventsListener(_threading.Thread):
 					_log.exception("processing %s", n)
 
 			elif self.tick_period:
-				idle_reads += 1
-				if idle_reads % _IDLE_READS == 0:
-					idle_reads = 0
+				idle_reads = (idle_reads + 1) % _IDLE_READS
+				if idle_reads == 0:
 					now = _timestamp()
 					if now - last_tick >= self.tick_period:
 						last_tick = now

lib/logitech/unifying_receiver/status.py

@@ -170,8 +170,8 @@ class DeviceStatus(dict):
 			# read these from the device in case they haven't been read already
 			# d.protocol, d.serial, d.firmware
 
-			# if BATTERY_LEVEL not in self:
-			self.read_battery(timestamp)
+			if BATTERY_LEVEL not in self:
+				self.read_battery(timestamp)
 
 			# make sure we know all the features of the device
 			if d.features:
@@ -192,6 +192,12 @@ class DeviceStatus(dict):
 		if n.sub_id >= 0x40:
 			return self._process_hidpp10_notification(n)
 
+		# some custom battery events for HID++ 1.0 devices
+		if n.sub_id in (0x07, 0x0D) and len(n.data) == 3 and n.data[2:3] == b'\x00':
+			# _log.debug("%s (%s) custom battery notification %s", self._device, self._device.protocol, n)
+			if self._device.protocol < 2:
+				return self._process_hidpp10_custom_notification(n)
+
 		# assuming 0x00 to 0x3F are feature (HID++ 2.0) notifications
 		try:
 			feature = self._device.features[n.sub_id]
@@ -201,6 +207,17 @@ class DeviceStatus(dict):
 
 		return self._process_feature_notification(n, feature)
 
+	def _process_hidpp10_custom_notification(self, n):
+		if n.sub_id == 0x07:
+			# TODO
+			return True
+
+		if n.sub_id == 0x0D:
+			# TODO
+			return True
+
+		_log.warn("%s: unrecognized %s", self._device, n)
+
 	def _process_hidpp10_notification(self, n):
 		if n.sub_id == 0x40:
 			if n.address == 0x02:

lib/solaar/listener.py

@@ -33,7 +33,7 @@ DUMMY_RECEIVER = _GHOST_DEVICE(0xFF, 'Solaar', None, 'Receiver not found.', 0)
 
 # how often to poll devices that haven't updated their statuses on their own
 # (through notifications)
-_POLL_TICK = 60  # seconds
+_POLL_TICK = 120  # seconds
 
 
 class ReceiverListener(_listener.EventsListener):
@@ -93,8 +93,6 @@ class ReceiverListener(_listener.EventsListener):
 						device.status, alert, reason or '')
 
 		if device.kind is None:
-			# print ("self.receiver: ", self.receiver, id(self.receiver))
-			# print ("device: ", device, id(device))
 			assert device == self.receiver
 			# the status of the receiver changed
 			self.status_changed_callback(device, alert, reason)