refactor: Introduce Hidpp20 class

Related #1097

lib/logitech_receiver/base.py

@@ -28,9 +28,8 @@ from time import time as _timestamp
 
 import hidapi as _hid
 
-from . import exceptions
+from . import exceptions, hidpp20
 from . import hidpp10_constants as _hidpp10_constants
-from . import hidpp20 as _hidpp20
 from . import hidpp20_constants as _hidpp20_constants
 from .base_usb import ALL as _RECEIVER_USB_IDS
 from .common import strhex as _strhex
@@ -38,6 +37,8 @@ from .descriptors import DEVICES as _DEVICES
 
 logger = logging.getLogger(__name__)
 
+_hidpp20 = hidpp20.Hidpp20()
+
 #
 #
 #

lib/logitech_receiver/device.py

@@ -31,6 +31,7 @@ from .settings_templates import check_feature_settings as _check_feature_setting
 logger = logging.getLogger(__name__)
 
 _hidpp10 = hidpp10.Hidpp10()
+_hidpp20 = hidpp20.Hidpp20()
 _R = hidpp10_constants.REGISTERS
 _IR = hidpp10_constants.INFO_SUBREGISTERS
 
@@ -43,7 +44,7 @@ class Device:
     def __init__(self, receiver, number, online, pairing_info=None, handle=None, device_info=None, setting_callback=None):
         assert receiver or device_info
         if receiver:
-            assert number > 0 and number <= 15  # some receivers have devices past their max # of devices
+            assert 0 < number <= 15  # some receivers have devices past their max # of devices
         self.number = number  # will be None at this point for directly connected devices
         self.online = online
         self.descriptor = None
@@ -142,7 +143,7 @@ class Device:
             if not self.online:  # be very defensive
                 self.ping()
             if self.online and self.protocol >= 2.0:
-                self._codename = hidpp20.get_friendly_name(self)
+                self._codename = _hidpp20.get_friendly_name(self)
                 if not self._codename:
                     self._codename = self.name.split(" ", 1)[0] if self.name else None
             if not self._codename and self.receiver:
@@ -162,11 +163,11 @@ class Device:
                 except exceptions.NoSuchDevice:
                     pass
             if self.online and self.protocol >= 2.0:
-                self._name = hidpp20.get_name(self)
+                self._name = _hidpp20.get_name(self)
         return self._name or self._codename or ("Unknown device %s" % (self.wpid or self.product_id))
 
     def get_ids(self):
-        ids = hidpp20.get_ids(self)
+        ids = _hidpp20.get_ids(self)
         if ids:
             self._unitId, self._modelId, self._tid_map = ids
             if logger.isEnabledFor(logging.INFO) and self._serial and self._serial != self._unitId:
@@ -193,14 +194,14 @@ class Device:
     @property
     def kind(self):
         if not self._kind and self.online and self.protocol >= 2.0:
-            self._kind = hidpp20.get_kind(self)
+            self._kind = _hidpp20.get_kind(self)
         return self._kind or "?"
 
     @property
     def firmware(self):
         if self._firmware is None and self.online:
             if self.protocol >= 2.0:
-                self._firmware = hidpp20.get_firmware(self)
+                self._firmware = _hidpp20.get_firmware(self)
             else:
                 self._firmware = _hidpp10.get_firmware(self)
         return self._firmware or ()
@@ -220,7 +221,7 @@ class Device:
     @property
     def polling_rate(self):
         if self.online and self.protocol >= 2.0:
-            rate = hidpp20.get_polling_rate(self)
+            rate = _hidpp20.get_polling_rate(self)
             self._polling_rate = rate if rate else self._polling_rate
         return self._polling_rate
 
@@ -234,14 +235,14 @@ class Device:
     def keys(self):
         if not self._keys:
             if self.online and self.protocol >= 2.0:
-                self._keys = hidpp20.get_keys(self) or ()
+                self._keys = _hidpp20.get_keys(self) or ()
         return self._keys
 
     @property
     def remap_keys(self):
         if self._remap_keys is None:
             if self.online and self.protocol >= 2.0:
-                self._remap_keys = hidpp20.get_remap_keys(self) or ()
+                self._remap_keys = _hidpp20.get_remap_keys(self) or ()
         return self._remap_keys
 
     @property
@@ -250,21 +251,21 @@ class Device:
             with self._gestures_lock:
                 if self._gestures is None:
                     if self.online and self.protocol >= 2.0:
-                        self._gestures = hidpp20.get_gestures(self) or ()
+                        self._gestures = _hidpp20.get_gestures(self) or ()
         return self._gestures
 
     @property
     def backlight(self):
         if self._backlight is None:
             if self.online and self.protocol >= 2.0:
-                self._backlight = hidpp20.get_backlight(self)
+                self._backlight = _hidpp20.get_backlight(self)
         return self._backlight
 
     @property
     def profiles(self):
         if self._profiles is None:
             if self.online and self.protocol >= 2.0:
-                self._profiles = hidpp20.get_profiles(self)
+                self._profiles = _hidpp20.get_profiles(self)
         return self._profiles
 
     @property
@@ -301,7 +302,7 @@ class Device:
 
     def set_configuration(self, configuration, no_reply=False):
         if self.online and self.protocol >= 2.0:
-            hidpp20.config_change(self, configuration, no_reply=no_reply)
+            _hidpp20.config_change(self, configuration, no_reply=no_reply)
 
     def reset(self, no_reply=False):
         self.set_configuration(0, no_reply)
@@ -320,7 +321,7 @@ class Device:
         else:
             battery_feature = self.persister.get("_battery", None) if self.persister else None
             if battery_feature != 0:
-                result = hidpp20.get_battery(self, battery_feature)
+                result = _hidpp20.get_battery(self, battery_feature)
                 try:
                     feature, level, next, status, voltage = result
                     if self.persister and battery_feature is None:

lib/logitech_receiver/hidpp20.py

@@ -41,6 +41,8 @@ from .i18n import _
 
 logger = logging.getLogger(__name__)
 
+KIND_MAP = {kind: _hidpp10_constants.DEVICE_KIND[str(kind)] for kind in DEVICE_KIND}
+
 
 def hexint_presenter(dumper, data):
     return dumper.represent_int(hex(data))
@@ -1351,157 +1353,6 @@ def feature_request(device, feature, function=0x00, *params, no_reply=False):
             return device.request((feature_index << 8) + (function & 0xFF), *params, no_reply=no_reply)
 
 
-def get_firmware(device):
-    """Reads a device's firmware info.
-
-    :returns: a list of FirmwareInfo tuples, ordered by firmware layer.
-    """
-    count = feature_request(device, FEATURE.DEVICE_FW_VERSION)
-    if count:
-        count = ord(count[:1])
-
-        fw = []
-        for index in range(0, count):
-            fw_info = feature_request(device, FEATURE.DEVICE_FW_VERSION, 0x10, index)
-            if fw_info:
-                level = ord(fw_info[:1]) & 0x0F
-                if level == 0 or level == 1:
-                    name, version_major, version_minor, build = _unpack("!3sBBH", fw_info[1:8])
-                    version = "%02X.%02X" % (version_major, version_minor)
-                    if build:
-                        version += ".B%04X" % build
-                    extras = fw_info[9:].rstrip(b"\x00") or None
-                    fw_info = _FirmwareInfo(FIRMWARE_KIND[level], name.decode("ascii"), version, extras)
-                elif level == FIRMWARE_KIND.Hardware:
-                    fw_info = _FirmwareInfo(FIRMWARE_KIND.Hardware, "", str(ord(fw_info[1:2])), None)
-                else:
-                    fw_info = _FirmwareInfo(FIRMWARE_KIND.Other, "", "", None)
-
-                fw.append(fw_info)
-                # if logger.isEnabledFor(logging.DEBUG):
-                #     logger.debug("device %d firmware %s", devnumber, fw_info)
-        return tuple(fw)
-
-
-def get_ids(device):
-    """Reads a device's ids (unit and model numbers)"""
-    ids = feature_request(device, FEATURE.DEVICE_FW_VERSION)
-    if ids:
-        unitId = ids[1:5]
-        modelId = ids[7:13]
-        transport_bits = ord(ids[6:7])
-        offset = 0
-        tid_map = {}
-        for transport, flag in [("btid", 0x1), ("btleid", 0x02), ("wpid", 0x04), ("usbid", 0x08)]:
-            if transport_bits & flag:
-                tid_map[transport] = modelId[offset : offset + 2].hex().upper()
-                offset = offset + 2
-        return (unitId.hex().upper(), modelId.hex().upper(), tid_map)
-
-
-KIND_MAP = {kind: _hidpp10_constants.DEVICE_KIND[str(kind)] for kind in DEVICE_KIND}
-
-
-def get_kind(device):
-    """Reads a device's type.
-
-    :see DEVICE_KIND:
-    :returns: a string describing the device type, or ``None`` if the device is
-    not available or does not support the ``DEVICE_NAME`` feature.
-    """
-    kind = feature_request(device, FEATURE.DEVICE_NAME, 0x20)
-    if kind:
-        kind = ord(kind[:1])
-        # if logger.isEnabledFor(logging.DEBUG):
-        #     logger.debug("device %d type %d = %s", devnumber, kind, DEVICE_KIND[kind])
-        return KIND_MAP[DEVICE_KIND[kind]]
-
-
-def get_name(device):
-    """Reads a device's name.
-
-    :returns: a string with the device name, or ``None`` if the device is not
-    available or does not support the ``DEVICE_NAME`` feature.
-    """
-    name_length = feature_request(device, FEATURE.DEVICE_NAME)
-    if name_length:
-        name_length = ord(name_length[:1])
-
-        name = b""
-        while len(name) < name_length:
-            fragment = feature_request(device, FEATURE.DEVICE_NAME, 0x10, len(name))
-            if fragment:
-                name += fragment[: name_length - len(name)]
-            else:
-                logger.error("failed to read whole name of %s (expected %d chars)", device, name_length)
-                return None
-
-        return name.decode("utf-8")
-
-
-def get_friendly_name(device):
-    """Reads a device's friendly name.
-
-    :returns: a string with the device name, or ``None`` if the device is not
-    available or does not support the ``DEVICE_NAME`` feature.
-    """
-    name_length = feature_request(device, FEATURE.DEVICE_FRIENDLY_NAME)
-    if name_length:
-        name_length = ord(name_length[:1])
-
-        name = b""
-        while len(name) < name_length:
-            fragment = feature_request(device, FEATURE.DEVICE_FRIENDLY_NAME, 0x10, len(name))
-            if fragment:
-                initial_null = 0 if fragment[0] else 1  # initial null actually seen on a device
-                name += fragment[initial_null : name_length + initial_null - len(name)]
-            else:
-                logger.error("failed to read whole name of %s (expected %d chars)", device, name_length)
-                return None
-
-        return name.decode("utf-8")
-
-
-def get_battery_status(device):
-    report = feature_request(device, FEATURE.BATTERY_STATUS)
-    if report:
-        return decipher_battery_status(report)
-
-
-def decipher_battery_status(report):
-    discharge, next, status = _unpack("!BBB", report[:3])
-    discharge = None if discharge == 0 else discharge
-    status = _BATTERY_STATUS[status]
-    if logger.isEnabledFor(logging.DEBUG):
-        logger.debug("battery status %s%% charged, next %s%%, status %s", discharge, next, status)
-    return FEATURE.BATTERY_STATUS, discharge, next, status, None
-
-
-def get_battery_unified(device):
-    report = feature_request(device, FEATURE.UNIFIED_BATTERY, 0x10)
-    if report is not None:
-        return decipher_battery_unified(report)
-
-
-def decipher_battery_unified(report):
-    discharge, level, status, _ignore = _unpack("!BBBB", report[:4])
-    status = _BATTERY_STATUS[status]
-    if logger.isEnabledFor(logging.DEBUG):
-        logger.debug("battery unified %s%% charged, level %s, charging %s", discharge, level, status)
-    level = (
-        _BATTERY_APPROX.full
-        if level == 8  # full
-        else _BATTERY_APPROX.good
-        if level == 4  # good
-        else _BATTERY_APPROX.low
-        if level == 2  # low
-        else _BATTERY_APPROX.critical
-        if level == 1  # critical
-        else _BATTERY_APPROX.empty
-    )
-    return FEATURE.UNIFIED_BATTERY, discharge if discharge else level, None, status, None
-
-
 # voltage to remaining charge from Logitech
 battery_voltage_remaining = (
     (4186, 100),
@@ -1521,10 +1372,366 @@ battery_voltage_remaining = (
 )
 
 
-def get_battery_voltage(device):
+class Hidpp20:
-    report = feature_request(device, FEATURE.BATTERY_VOLTAGE)
+    def get_firmware(self, device):
-    if report is not None:
+        """Reads a device's firmware info.
-        return decipher_battery_voltage(report)
+
+        :returns: a list of FirmwareInfo tuples, ordered by firmware layer.
+        """
+        count = feature_request(device, FEATURE.DEVICE_FW_VERSION)
+        if count:
+            count = ord(count[:1])
+
+            fw = []
+            for index in range(0, count):
+                fw_info = feature_request(device, FEATURE.DEVICE_FW_VERSION, 0x10, index)
+                if fw_info:
+                    level = ord(fw_info[:1]) & 0x0F
+                    if level == 0 or level == 1:
+                        name, version_major, version_minor, build = _unpack("!3sBBH", fw_info[1:8])
+                        version = "%02X.%02X" % (version_major, version_minor)
+                        if build:
+                            version += ".B%04X" % build
+                        extras = fw_info[9:].rstrip(b"\x00") or None
+                        fw_info = _FirmwareInfo(FIRMWARE_KIND[level], name.decode("ascii"), version, extras)
+                    elif level == FIRMWARE_KIND.Hardware:
+                        fw_info = _FirmwareInfo(FIRMWARE_KIND.Hardware, "", str(ord(fw_info[1:2])), None)
+                    else:
+                        fw_info = _FirmwareInfo(FIRMWARE_KIND.Other, "", "", None)
+
+                    fw.append(fw_info)
+                    # if logger.isEnabledFor(logging.DEBUG):
+                    #     logger.debug("device %d firmware %s", devnumber, fw_info)
+            return tuple(fw)
+
+    def get_ids(self, device):
+        """Reads a device's ids (unit and model numbers)"""
+        ids = feature_request(device, FEATURE.DEVICE_FW_VERSION)
+        if ids:
+            unitId = ids[1:5]
+            modelId = ids[7:13]
+            transport_bits = ord(ids[6:7])
+            offset = 0
+            tid_map = {}
+            for transport, flag in [("btid", 0x1), ("btleid", 0x02), ("wpid", 0x04), ("usbid", 0x08)]:
+                if transport_bits & flag:
+                    tid_map[transport] = modelId[offset : offset + 2].hex().upper()
+                    offset = offset + 2
+            return (unitId.hex().upper(), modelId.hex().upper(), tid_map)
+
+    def get_kind(self, device):
+        """Reads a device's type.
+
+        :see DEVICE_KIND:
+        :returns: a string describing the device type, or ``None`` if the device is
+        not available or does not support the ``DEVICE_NAME`` feature.
+        """
+        kind = feature_request(device, FEATURE.DEVICE_NAME, 0x20)
+        if kind:
+            kind = ord(kind[:1])
+            # if logger.isEnabledFor(logging.DEBUG):
+            #     logger.debug("device %d type %d = %s", devnumber, kind, DEVICE_KIND[kind])
+            return KIND_MAP[DEVICE_KIND[kind]]
+
+    def get_name(self, device):
+        """Reads a device's name.
+
+        :returns: a string with the device name, or ``None`` if the device is not
+        available or does not support the ``DEVICE_NAME`` feature.
+        """
+        name_length = feature_request(device, FEATURE.DEVICE_NAME)
+        if name_length:
+            name_length = ord(name_length[:1])
+
+            name = b""
+            while len(name) < name_length:
+                fragment = feature_request(device, FEATURE.DEVICE_NAME, 0x10, len(name))
+                if fragment:
+                    name += fragment[: name_length - len(name)]
+                else:
+                    logger.error("failed to read whole name of %s (expected %d chars)", device, name_length)
+                    return None
+
+            return name.decode("utf-8")
+
+    def get_friendly_name(self, device):
+        """Reads a device's friendly name.
+
+        :returns: a string with the device name, or ``None`` if the device is not
+        available or does not support the ``DEVICE_NAME`` feature.
+        """
+        name_length = feature_request(device, FEATURE.DEVICE_FRIENDLY_NAME)
+        if name_length:
+            name_length = ord(name_length[:1])
+
+            name = b""
+            while len(name) < name_length:
+                fragment = feature_request(device, FEATURE.DEVICE_FRIENDLY_NAME, 0x10, len(name))
+                if fragment:
+                    initial_null = 0 if fragment[0] else 1  # initial null actually seen on a device
+                    name += fragment[initial_null : name_length + initial_null - len(name)]
+                else:
+                    logger.error("failed to read whole name of %s (expected %d chars)", device, name_length)
+                    return None
+
+            return name.decode("utf-8")
+
+    def get_battery_status(self, device):
+        report = feature_request(device, FEATURE.BATTERY_STATUS)
+        if report:
+            return decipher_battery_status(report)
+
+    def get_battery_unified(self, device):
+        report = feature_request(device, FEATURE.UNIFIED_BATTERY, 0x10)
+        if report is not None:
+            return decipher_battery_unified(report)
+
+    def get_battery_voltage(self, device):
+        report = feature_request(device, FEATURE.BATTERY_VOLTAGE)
+        if report is not None:
+            return decipher_battery_voltage(report)
+
+    def get_adc_measurement(self, device):
+        try:  # this feature call produces an error for headsets that are connected but inactive
+            report = feature_request(device, FEATURE.ADC_MEASUREMENT)
+            if report is not None:
+                return decipher_adc_measurement(report)
+        except exceptions.FeatureCallError:
+            return FEATURE.ADC_MEASUREMENT if FEATURE.ADC_MEASUREMENT in device.features else None
+
+    def get_battery(self, device, feature):
+        """Return battery information - feature, approximate level, next, charging, voltage
+        or battery feature if there is one but it is not responding or None for no battery feature"""
+
+        if feature is not None:
+            battery_function = battery_functions.get(feature, None)
+            if battery_function:
+                result = battery_function(self, device)
+                if result:
+                    return result
+        else:
+            for battery_function in battery_functions.values():
+                result = battery_function(self, device)
+                if result:
+                    return result
+        return 0
+
+    def get_keys(self, device):
+        # TODO: add here additional variants for other REPROG_CONTROLS
+        count = None
+        if FEATURE.REPROG_CONTROLS_V2 in device.features:
+            count = feature_request(device, FEATURE.REPROG_CONTROLS_V2)
+            return KeysArrayV1(device, ord(count[:1]))
+        elif FEATURE.REPROG_CONTROLS_V4 in device.features:
+            count = feature_request(device, FEATURE.REPROG_CONTROLS_V4)
+            return KeysArrayV4(device, ord(count[:1]))
+        return None
+
+    def get_remap_keys(self, device):
+        count = feature_request(device, FEATURE.PERSISTENT_REMAPPABLE_ACTION, 0x10)
+        if count:
+            return KeysArrayPersistent(device, ord(count[:1]))
+
+    def get_gestures(self, device):
+        if getattr(device, "_gestures", None) is not None:
+            return device._gestures
+        if FEATURE.GESTURE_2 in device.features:
+            return Gestures(device)
+
+    def get_backlight(self, device):
+        if getattr(device, "_backlight", None) is not None:
+            return device._backlight
+        if FEATURE.BACKLIGHT2 in device.features:
+            return Backlight(device)
+
+    def get_profiles(self, device):
+        if getattr(device, "_profiles", None) is not None:
+            return device._profiles
+        if FEATURE.ONBOARD_PROFILES in device.features:
+            return OnboardProfiles.from_device(device)
+
+    def get_mouse_pointer_info(self, device):
+        pointer_info = feature_request(device, FEATURE.MOUSE_POINTER)
+        if pointer_info:
+            dpi, flags = _unpack("!HB", pointer_info[:3])
+            acceleration = ("none", "low", "med", "high")[flags & 0x3]
+            suggest_os_ballistics = (flags & 0x04) != 0
+            suggest_vertical_orientation = (flags & 0x08) != 0
+            return {
+                "dpi": dpi,
+                "acceleration": acceleration,
+                "suggest_os_ballistics": suggest_os_ballistics,
+                "suggest_vertical_orientation": suggest_vertical_orientation,
+            }
+
+    def get_vertical_scrolling_info(self, device):
+        vertical_scrolling_info = feature_request(device, FEATURE.VERTICAL_SCROLLING)
+        if vertical_scrolling_info:
+            roller, ratchet, lines = _unpack("!BBB", vertical_scrolling_info[:3])
+            roller_type = (
+                "reserved",
+                "standard",
+                "reserved",
+                "3G",
+                "micro",
+                "normal touch pad",
+                "inverted touch pad",
+                "reserved",
+            )[roller]
+            return {"roller": roller_type, "ratchet": ratchet, "lines": lines}
+
+    def get_hi_res_scrolling_info(self, device):
+        hi_res_scrolling_info = feature_request(device, FEATURE.HI_RES_SCROLLING)
+        if hi_res_scrolling_info:
+            mode, resolution = _unpack("!BB", hi_res_scrolling_info[:2])
+            return mode, resolution
+
+    def get_pointer_speed_info(self, device):
+        pointer_speed_info = feature_request(device, FEATURE.POINTER_SPEED)
+        if pointer_speed_info:
+            pointer_speed_hi, pointer_speed_lo = _unpack("!BB", pointer_speed_info[:2])
+            # if pointer_speed_lo > 0:
+            #     pointer_speed_lo = pointer_speed_lo
+            return pointer_speed_hi + pointer_speed_lo / 256
+
+    def get_lowres_wheel_status(self, device):
+        lowres_wheel_status = feature_request(device, FEATURE.LOWRES_WHEEL)
+        if lowres_wheel_status:
+            wheel_flag = _unpack("!B", lowres_wheel_status[:1])[0]
+            wheel_reporting = ("HID", "HID++")[wheel_flag & 0x01]
+            return wheel_reporting
+
+    def get_hires_wheel(self, device):
+        caps = feature_request(device, FEATURE.HIRES_WHEEL, 0x00)
+        mode = feature_request(device, FEATURE.HIRES_WHEEL, 0x10)
+        ratchet = feature_request(device, FEATURE.HIRES_WHEEL, 0x030)
+
+        if caps and mode and ratchet:
+            # Parse caps
+            multi, flags = _unpack("!BB", caps[:2])
+
+            has_invert = (flags & 0x08) != 0
+            has_ratchet = (flags & 0x04) != 0
+
+            # Parse mode
+            wheel_mode, reserved = _unpack("!BB", mode[:2])
+
+            target = (wheel_mode & 0x01) != 0
+            res = (wheel_mode & 0x02) != 0
+            inv = (wheel_mode & 0x04) != 0
+
+            # Parse Ratchet switch
+            ratchet_mode, reserved = _unpack("!BB", ratchet[:2])
+
+            ratchet = (ratchet_mode & 0x01) != 0
+
+            return multi, has_invert, has_ratchet, inv, res, target, ratchet
+
+    def get_new_fn_inversion(self, device):
+        state = feature_request(device, FEATURE.NEW_FN_INVERSION, 0x00)
+        if state:
+            inverted, default_inverted = _unpack("!BB", state[:2])
+            inverted = (inverted & 0x01) != 0
+            default_inverted = (default_inverted & 0x01) != 0
+            return inverted, default_inverted
+
+    def get_host_names(self, device):
+        state = feature_request(device, FEATURE.HOSTS_INFO, 0x00)
+        host_names = {}
+        if state:
+            capability_flags, _ignore, numHosts, currentHost = _unpack("!BBBB", state[:4])
+            if capability_flags & 0x01:  # device can get host names
+                for host in range(0, numHosts):
+                    hostinfo = feature_request(device, FEATURE.HOSTS_INFO, 0x10, host)
+                    _ignore, status, _ignore, _ignore, nameLen, _ignore = _unpack("!BBBBBB", hostinfo[:6])
+                    name = ""
+                    remaining = nameLen
+                    while remaining > 0:
+                        name_piece = feature_request(device, FEATURE.HOSTS_INFO, 0x30, host, nameLen - remaining)
+                        if name_piece:
+                            name += name_piece[2 : 2 + min(remaining, 14)].decode()
+                            remaining = max(0, remaining - 14)
+                        else:
+                            remaining = 0
+                    host_names[host] = (bool(status), name)
+            if host_names:  # update the current host's name if it doesn't match the system name
+                hostname = socket.gethostname().partition(".")[0]
+                if host_names[currentHost][1] != hostname:
+                    self.set_host_name(device, hostname, host_names[currentHost][1])
+                    host_names[currentHost] = (host_names[currentHost][0], hostname)
+        return host_names
+
+    def set_host_name(self, device, name, currentName=""):
+        name = bytearray(name, "utf-8")
+        currentName = bytearray(currentName, "utf-8")
+        if logger.isEnabledFor(logging.INFO):
+            logger.info("Setting host name to %s", name)
+        state = feature_request(device, FEATURE.HOSTS_INFO, 0x00)
+        if state:
+            flags, _ignore, _ignore, currentHost = _unpack("!BBBB", state[:4])
+            if flags & 0x02:
+                hostinfo = feature_request(device, FEATURE.HOSTS_INFO, 0x10, currentHost)
+                _ignore, _ignore, _ignore, _ignore, _ignore, maxNameLen = _unpack("!BBBBBB", hostinfo[:6])
+                if name[:maxNameLen] == currentName[:maxNameLen] and False:
+                    return True
+                length = min(maxNameLen, len(name))
+                chunk = 0
+                while chunk < length:
+                    response = feature_request(device, FEATURE.HOSTS_INFO, 0x40, currentHost, chunk, name[chunk : chunk + 14])
+                    if not response:
+                        return False
+                    chunk += 14
+            return True
+
+    def get_onboard_mode(self, device):
+        state = feature_request(device, FEATURE.ONBOARD_PROFILES, 0x20)
+
+        if state:
+            mode = _unpack("!B", state[:1])[0]
+            return mode
+
+    def set_onboard_mode(self, device, mode):
+        state = feature_request(device, FEATURE.ONBOARD_PROFILES, 0x10, mode)
+        return state
+
+    def get_polling_rate(self, device):
+        state = feature_request(device, FEATURE.REPORT_RATE, 0x10)
+        if state:
+            rate = _unpack("!B", state[:1])[0]
+            return str(rate) + "ms"
+        else:
+            rates = ["8ms", "4ms", "2ms", "1ms", "500us", "250us", "125us"]
+            state = feature_request(device, FEATURE.EXTENDED_ADJUSTABLE_REPORT_RATE, 0x20)
+            if state:
+                rate = _unpack("!B", state[:1])[0]
+                return rates[rate]
+
+    def get_remaining_pairing(self, device):
+        result = feature_request(device, FEATURE.REMAINING_PAIRING, 0x0)
+        if result:
+            result = _unpack("!B", result[:1])[0]
+            FEATURE._fallback = lambda x: "unknown:%04X" % x
+            return result
+
+    def config_change(self, device, configuration, no_reply=False):
+        return feature_request(device, FEATURE.CONFIG_CHANGE, 0x00, configuration, no_reply=no_reply)
+
+
+battery_functions = {
+    FEATURE.BATTERY_STATUS: Hidpp20.get_battery_status,
+    FEATURE.BATTERY_VOLTAGE: Hidpp20.get_battery_voltage,
+    FEATURE.UNIFIED_BATTERY: Hidpp20.get_battery_unified,
+    FEATURE.ADC_MEASUREMENT: Hidpp20.get_adc_measurement,
+}
+
+
+def decipher_battery_status(report):
+    discharge, next, status = _unpack("!BBB", report[:3])
+    discharge = None if discharge == 0 else discharge
+    status = _BATTERY_STATUS[status]
+    if logger.isEnabledFor(logging.DEBUG):
+        logger.debug("battery status %s%% charged, next %s%%, status %s", discharge, next, status)
+    return FEATURE.BATTERY_STATUS, discharge, next, status, None
 
 
 def decipher_battery_voltage(report):
@@ -1565,13 +1772,23 @@ def decipher_battery_voltage(report):
     return FEATURE.BATTERY_VOLTAGE, charge_lvl, None, status, voltage
 
 
-def get_adc_measurement(device):
+def decipher_battery_unified(report):
-    try:  # this feature call produces an error for headsets that are connected but inactive
+    discharge, level, status, _ignore = _unpack("!BBBB", report[:4])
-        report = feature_request(device, FEATURE.ADC_MEASUREMENT)
+    status = _BATTERY_STATUS[status]
-        if report is not None:
+    if logger.isEnabledFor(logging.DEBUG):
-            return decipher_adc_measurement(report)
+        logger.debug("battery unified %s%% charged, level %s, charging %s", discharge, level, status)
-    except exceptions.FeatureCallError:
+    level = (
-        return FEATURE.ADC_MEASUREMENT if FEATURE.ADC_MEASUREMENT in device.features else None
+        _BATTERY_APPROX.full
+        if level == 8  # full
+        else _BATTERY_APPROX.good
+        if level == 4  # good
+        else _BATTERY_APPROX.low
+        if level == 2  # low
+        else _BATTERY_APPROX.critical
+        if level == 1  # critical
+        else _BATTERY_APPROX.empty
+    )
+    return FEATURE.UNIFIED_BATTERY, discharge if discharge else level, None, status, None
 
 
 def decipher_adc_measurement(report):
@@ -1584,247 +1801,3 @@ def decipher_adc_measurement(report):
     if flags & 0x01:
         status = _BATTERY_STATUS.recharging if flags & 0x02 else _BATTERY_STATUS.discharging
         return FEATURE.ADC_MEASUREMENT, charge_level, None, status, adc
-
-
-battery_functions = {
-    FEATURE.BATTERY_STATUS: get_battery_status,
-    FEATURE.BATTERY_VOLTAGE: get_battery_voltage,
-    FEATURE.UNIFIED_BATTERY: get_battery_unified,
-    FEATURE.ADC_MEASUREMENT: get_adc_measurement,
-}
-
-
-def get_battery(device, feature):
-    """Return battery information - feature, approximate level, next, charging, voltage
-    or battery feature if there is one but it is not responding or None for no battery feature"""
-    if feature is not None:
-        battery_function = battery_functions.get(feature, None)
-        if battery_function:
-            result = battery_function(device)
-            if result:
-                return result
-    else:
-        for battery_function in battery_functions.values():
-            result = battery_function(device)
-            if result:
-                return result
-    return 0
-
-
-def get_keys(device):
-    # TODO: add here additional variants for other REPROG_CONTROLS
-    count = None
-    if FEATURE.REPROG_CONTROLS_V2 in device.features:
-        count = feature_request(device, FEATURE.REPROG_CONTROLS_V2)
-        return KeysArrayV1(device, ord(count[:1]))
-    elif FEATURE.REPROG_CONTROLS_V4 in device.features:
-        count = feature_request(device, FEATURE.REPROG_CONTROLS_V4)
-        return KeysArrayV4(device, ord(count[:1]))
-    return None
-
-
-def get_remap_keys(device):
-    count = feature_request(device, FEATURE.PERSISTENT_REMAPPABLE_ACTION, 0x10)
-    if count:
-        return KeysArrayPersistent(device, ord(count[:1]))
-
-
-def get_gestures(device):
-    if getattr(device, "_gestures", None) is not None:
-        return device._gestures
-    if FEATURE.GESTURE_2 in device.features:
-        return Gestures(device)
-
-
-def get_backlight(device):
-    if getattr(device, "_backlight", None) is not None:
-        return device._backlight
-    if FEATURE.BACKLIGHT2 in device.features:
-        return Backlight(device)
-
-
-def get_profiles(device):
-    if getattr(device, "_profiles", None) is not None:
-        return device._profiles
-    if FEATURE.ONBOARD_PROFILES in device.features:
-        return OnboardProfiles.from_device(device)
-
-
-def get_mouse_pointer_info(device):
-    pointer_info = feature_request(device, FEATURE.MOUSE_POINTER)
-    if pointer_info:
-        dpi, flags = _unpack("!HB", pointer_info[:3])
-        acceleration = ("none", "low", "med", "high")[flags & 0x3]
-        suggest_os_ballistics = (flags & 0x04) != 0
-        suggest_vertical_orientation = (flags & 0x08) != 0
-        return {
-            "dpi": dpi,
-            "acceleration": acceleration,
-            "suggest_os_ballistics": suggest_os_ballistics,
-            "suggest_vertical_orientation": suggest_vertical_orientation,
-        }
-
-
-def get_vertical_scrolling_info(device):
-    vertical_scrolling_info = feature_request(device, FEATURE.VERTICAL_SCROLLING)
-    if vertical_scrolling_info:
-        roller, ratchet, lines = _unpack("!BBB", vertical_scrolling_info[:3])
-        roller_type = (
-            "reserved",
-            "standard",
-            "reserved",
-            "3G",
-            "micro",
-            "normal touch pad",
-            "inverted touch pad",
-            "reserved",
-        )[roller]
-        return {"roller": roller_type, "ratchet": ratchet, "lines": lines}
-
-
-def get_hi_res_scrolling_info(device):
-    hi_res_scrolling_info = feature_request(device, FEATURE.HI_RES_SCROLLING)
-    if hi_res_scrolling_info:
-        mode, resolution = _unpack("!BB", hi_res_scrolling_info[:2])
-        return mode, resolution
-
-
-def get_pointer_speed_info(device):
-    pointer_speed_info = feature_request(device, FEATURE.POINTER_SPEED)
-    if pointer_speed_info:
-        pointer_speed_hi, pointer_speed_lo = _unpack("!BB", pointer_speed_info[:2])
-        # if pointer_speed_lo > 0:
-        #     pointer_speed_lo = pointer_speed_lo
-        return pointer_speed_hi + pointer_speed_lo / 256
-
-
-def get_lowres_wheel_status(device):
-    lowres_wheel_status = feature_request(device, FEATURE.LOWRES_WHEEL)
-    if lowres_wheel_status:
-        wheel_flag = _unpack("!B", lowres_wheel_status[:1])[0]
-        wheel_reporting = ("HID", "HID++")[wheel_flag & 0x01]
-        return wheel_reporting
-
-
-def get_hires_wheel(device):
-    caps = feature_request(device, FEATURE.HIRES_WHEEL, 0x00)
-    mode = feature_request(device, FEATURE.HIRES_WHEEL, 0x10)
-    ratchet = feature_request(device, FEATURE.HIRES_WHEEL, 0x030)
-
-    if caps and mode and ratchet:
-        # Parse caps
-        multi, flags = _unpack("!BB", caps[:2])
-
-        has_invert = (flags & 0x08) != 0
-        has_ratchet = (flags & 0x04) != 0
-
-        # Parse mode
-        wheel_mode, reserved = _unpack("!BB", mode[:2])
-
-        target = (wheel_mode & 0x01) != 0
-        res = (wheel_mode & 0x02) != 0
-        inv = (wheel_mode & 0x04) != 0
-
-        # Parse Ratchet switch
-        ratchet_mode, reserved = _unpack("!BB", ratchet[:2])
-
-        ratchet = (ratchet_mode & 0x01) != 0
-
-        return multi, has_invert, has_ratchet, inv, res, target, ratchet
-
-
-def get_new_fn_inversion(device):
-    state = feature_request(device, FEATURE.NEW_FN_INVERSION, 0x00)
-    if state:
-        inverted, default_inverted = _unpack("!BB", state[:2])
-        inverted = (inverted & 0x01) != 0
-        default_inverted = (default_inverted & 0x01) != 0
-        return inverted, default_inverted
-
-
-def get_host_names(device):
-    state = feature_request(device, FEATURE.HOSTS_INFO, 0x00)
-    host_names = {}
-    if state:
-        capability_flags, _ignore, numHosts, currentHost = _unpack("!BBBB", state[:4])
-        if capability_flags & 0x01:  # device can get host names
-            for host in range(0, numHosts):
-                hostinfo = feature_request(device, FEATURE.HOSTS_INFO, 0x10, host)
-                _ignore, status, _ignore, _ignore, nameLen, _ignore = _unpack("!BBBBBB", hostinfo[:6])
-                name = ""
-                remaining = nameLen
-                while remaining > 0:
-                    name_piece = feature_request(device, FEATURE.HOSTS_INFO, 0x30, host, nameLen - remaining)
-                    if name_piece:
-                        name += name_piece[2 : 2 + min(remaining, 14)].decode()
-                        remaining = max(0, remaining - 14)
-                    else:
-                        remaining = 0
-                host_names[host] = (bool(status), name)
-        if host_names:  # update the current host's name if it doesn't match the system name
-            hostname = socket.gethostname().partition(".")[0]
-            if host_names[currentHost][1] != hostname:
-                set_host_name(device, hostname, host_names[currentHost][1])
-                host_names[currentHost] = (host_names[currentHost][0], hostname)
-    return host_names
-
-
-def set_host_name(device, name, currentName=""):
-    name = bytearray(name, "utf-8")
-    currentName = bytearray(currentName, "utf-8")
-    if logger.isEnabledFor(logging.INFO):
-        logger.info("Setting host name to %s", name)
-    state = feature_request(device, FEATURE.HOSTS_INFO, 0x00)
-    if state:
-        flags, _ignore, _ignore, currentHost = _unpack("!BBBB", state[:4])
-        if flags & 0x02:
-            hostinfo = feature_request(device, FEATURE.HOSTS_INFO, 0x10, currentHost)
-            _ignore, _ignore, _ignore, _ignore, _ignore, maxNameLen = _unpack("!BBBBBB", hostinfo[:6])
-            if name[:maxNameLen] == currentName[:maxNameLen] and False:
-                return True
-            length = min(maxNameLen, len(name))
-            chunk = 0
-            while chunk < length:
-                response = feature_request(device, FEATURE.HOSTS_INFO, 0x40, currentHost, chunk, name[chunk : chunk + 14])
-                if not response:
-                    return False
-                chunk += 14
-        return True
-
-
-def get_onboard_mode(device):
-    state = feature_request(device, FEATURE.ONBOARD_PROFILES, 0x20)
-
-    if state:
-        mode = _unpack("!B", state[:1])[0]
-        return mode
-
-
-def set_onboard_mode(device, mode):
-    state = feature_request(device, FEATURE.ONBOARD_PROFILES, 0x10, mode)
-    return state
-
-
-def get_polling_rate(device):
-    state = feature_request(device, FEATURE.REPORT_RATE, 0x10)
-    if state:
-        rate = _unpack("!B", state[:1])[0]
-        return str(rate) + "ms"
-    else:
-        rates = ["8ms", "4ms", "2ms", "1ms", "500us", "250us", "125us"]
-        state = feature_request(device, FEATURE.EXTENDED_ADJUSTABLE_REPORT_RATE, 0x20)
-        if state:
-            rate = _unpack("!B", state[:1])[0]
-            return rates[rate]
-
-
-def get_remaining_pairing(device):
-    result = feature_request(device, FEATURE.REMAINING_PAIRING, 0x0)
-    if result:
-        result = _unpack("!B", result[:1])[0]
-        FEATURE._fallback = lambda x: "unknown:%04X" % x
-        return result
-
-
-def config_change(device, configuration, no_reply=False):
-    return feature_request(device, FEATURE.CONFIG_CHANGE, 0x00, configuration, no_reply=no_reply)

lib/logitech_receiver/notifications.py

@@ -24,9 +24,8 @@ import threading as _threading
 from struct import unpack as _unpack
 
 from . import diversion as _diversion
-from . import hidpp10
+from . import hidpp10, hidpp20
 from . import hidpp10_constants as _hidpp10_constants
-from . import hidpp20 as _hidpp20
 from . import hidpp20_constants as _hidpp20_constants
 from . import settings_templates as _st
 from .base import DJ_MESSAGE_ID as _DJ_MESSAGE_ID
@@ -39,6 +38,7 @@ from .status import KEYS as _K
 logger = logging.getLogger(__name__)
 
 _hidpp10 = hidpp10.Hidpp10()
+_hidpp20 = hidpp20.Hidpp20()
 _R = _hidpp10_constants.REGISTERS
 _F = _hidpp20_constants.FEATURE
 
@@ -296,7 +296,7 @@ def _process_feature_notification(device, status, n, feature):
 
     if feature == _F.BATTERY_STATUS:
         if n.address == 0x00:
-            _ignore, discharge_level, discharge_next_level, battery_status, voltage = _hidpp20.decipher_battery_status(n.data)
+            _ignore, discharge_level, discharge_next_level, battery_status, voltage = hidpp20.decipher_battery_status(n.data)
             status.set_battery_info(discharge_level, discharge_next_level, battery_status, voltage)
         elif n.address == 0x10:
             if logger.isEnabledFor(logging.INFO):
@@ -306,21 +306,21 @@ def _process_feature_notification(device, status, n, feature):
 
     elif feature == _F.BATTERY_VOLTAGE:
         if n.address == 0x00:
-            _ignore, level, nextl, battery_status, voltage = _hidpp20.decipher_battery_voltage(n.data)
+            _ignore, level, nextl, battery_status, voltage = hidpp20.decipher_battery_voltage(n.data)
             status.set_battery_info(level, nextl, battery_status, voltage)
         else:
             logger.warning("%s: unknown VOLTAGE %s", device, n)
 
     elif feature == _F.UNIFIED_BATTERY:
         if n.address == 0x00:
-            _ignore, level, nextl, battery_status, voltage = _hidpp20.decipher_battery_unified(n.data)
+            _ignore, level, nextl, battery_status, voltage = hidpp20.decipher_battery_unified(n.data)
             status.set_battery_info(level, nextl, battery_status, voltage)
         else:
             logger.warning("%s: unknown UNIFIED BATTERY %s", device, n)
 
     elif feature == _F.ADC_MEASUREMENT:
         if n.address == 0x00:
-            result = _hidpp20.decipher_adc_measurement(n.data)
+            result = hidpp20.decipher_adc_measurement(n.data)
             if result:
                 _ignore, level, nextl, battery_status, voltage = result
                 status.set_battery_info(level, nextl, battery_status, voltage)

lib/logitech_receiver/settings_templates.py

@@ -25,7 +25,7 @@ from traceback import format_exc as _format_exc
 
 from . import descriptors as _descriptors
 from . import hidpp10_constants as _hidpp10_constants
-from . import hidpp20 as _hidpp20
+from . import hidpp20
 from . import hidpp20_constants as _hidpp20_constants
 from . import notify as _notify
 from . import special_keys as _special_keys
@@ -58,6 +58,7 @@ from .special_keys import DISABLE as _DKEY
 
 logger = logging.getLogger(__name__)
 
+_hidpp20 = hidpp20.Hidpp20()
 _DK = _hidpp10_constants.DEVICE_KIND
 _R = _hidpp10_constants.REGISTERS
 _F = _hidpp20_constants.FEATURE
@@ -526,7 +527,7 @@ class OnboardProfiles(_Setting):
     class validator_class(_ChoicesV):
         @classmethod
         def build(cls, setting_class, device):
-            headers = _hidpp20.OnboardProfiles.get_profile_headers(device)
+            headers = hidpp20.OnboardProfiles.get_profile_headers(device)
             profiles_list = [setting_class.choices_universe[0]]
             if headers:
                 for sector, enabled in headers:
@@ -1232,7 +1233,7 @@ class Gesture2Gestures(_BitFieldOMSetting):
     description = _("Tweak the mouse/touchpad behaviour.")
     feature = _F.GESTURE_2
     rw_options = {"read_fnid": 0x10, "write_fnid": 0x20}
-    validator_options = {"om_method": _hidpp20.Gesture.enable_offset_mask}
+    validator_options = {"om_method": hidpp20.Gesture.enable_offset_mask}
     choices_universe = _hidpp20_constants.GESTURE
     _labels = _GESTURE2_GESTURES_LABELS
 
@@ -1249,7 +1250,7 @@ class Gesture2Divert(_BitFieldOMSetting):
     description = _("Divert mouse/touchpad gestures.")
     feature = _F.GESTURE_2
     rw_options = {"read_fnid": 0x30, "write_fnid": 0x40}
-    validator_options = {"om_method": _hidpp20.Gesture.diversion_offset_mask}
+    validator_options = {"om_method": hidpp20.Gesture.diversion_offset_mask}
     choices_universe = _hidpp20_constants.GESTURE
     _labels = _GESTURE2_GESTURES_LABELS
 
@@ -1266,8 +1267,8 @@ class Gesture2Params(_LongSettings):
     description = _("Change numerical parameters of a mouse/touchpad.")
     feature = _F.GESTURE_2
     rw_options = {"read_fnid": 0x70, "write_fnid": 0x80}
-    choices_universe = _hidpp20.PARAM
+    choices_universe = hidpp20.PARAM
-    sub_items_universe = _hidpp20.SUB_PARAM
+    sub_items_universe = hidpp20.SUB_PARAM
     # item (NamedInt) -> list/tuple of objects that have the following attributes
     # .id (sub-item text), .length (in bytes), .minimum and .maximum
 
@@ -1463,7 +1464,7 @@ class LEDControl(_Setting):
 
 
 colors = _special_keys.COLORS
-_LEDP = _hidpp20.LEDParam
+_LEDP = hidpp20.LEDParam
 
 
 # an LED Zone has an index, a set of possible LED effects, and an LED effect setting
@@ -1476,7 +1477,7 @@ class LEDZoneSetting(_Setting):
     speed_field = {"name": _LEDP.speed, "kind": _KIND.range, "label": _("Speed"), "min": 0, "max": 255}
     period_field = {"name": _LEDP.period, "kind": _KIND.range, "label": _("Period"), "min": 100, "max": 5000}
     intensity_field = {"name": _LEDP.intensity, "kind": _KIND.range, "label": _("Intensity"), "min": 0, "max": 100}
-    ramp_field = {"name": _LEDP.ramp, "kind": _KIND.choice, "label": _("Ramp"), "choices": _hidpp20.LEDRampChoices}
+    ramp_field = {"name": _LEDP.ramp, "kind": _KIND.choice, "label": _("Ramp"), "choices": hidpp20.LEDRampChoices}
     #    form_field = { 'name': _LEDP.form, 'kind': _KIND.choice, 'label': _('Form'), 'choices': _hidpp20.LEDFormChoices }
     possible_fields = [color_field, speed_field, period_field, intensity_field, ramp_field]
 
@@ -1487,14 +1488,14 @@ class LEDZoneSetting(_Setting):
         for zone in infos.zones:
             prefix = _int2bytes(zone.index, 1)
             rw = _FeatureRW(_F.COLOR_LED_EFFECTS, read_fnid=0xE0, write_fnid=0x30, prefix=prefix)
-            validator = _HeteroV(data_class=_hidpp20.LEDEffectSetting, options=zone.effects, readable=infos.readable)
+            validator = _HeteroV(data_class=hidpp20.LEDEffectSetting, options=zone.effects, readable=infos.readable)
             setting = cls(device, rw, validator)
             setting.name = cls.name + str(int(zone.location))
-            setting.label = _("LEDs") + " " + str(_hidpp20.LEDZoneLocations[zone.location])
+            setting.label = _("LEDs") + " " + str(hidpp20.LEDZoneLocations[zone.location])
-            choices = [_hidpp20.LEDEffects[e.ID][0] for e in zone.effects]
+            choices = [hidpp20.LEDEffects[e.ID][0] for e in zone.effects]
             ID_field = {"name": "ID", "kind": _KIND.choice, "label": None, "choices": choices}
             setting.possible_fields = [ID_field] + cls.possible_fields
-            setting.fields_map = _hidpp20.LEDEffects
+            setting.fields_map = hidpp20.LEDEffects
             settings.append(setting)
         return settings
 

lib/solaar/cli/show.py

@@ -14,9 +14,8 @@
 ## with this program; if not, write to the Free Software Foundation, Inc.,
 ## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 
-from logitech_receiver import exceptions, hidpp10
+from logitech_receiver import exceptions, hidpp10, hidpp20
 from logitech_receiver import hidpp10_constants as _hidpp10_constants
-from logitech_receiver import hidpp20 as _hidpp20
 from logitech_receiver import hidpp20_constants as _hidpp20_constants
 from logitech_receiver import receiver as _receiver
 from logitech_receiver import settings_templates as _settings_templates
@@ -26,6 +25,7 @@ from logitech_receiver.common import strhex as _strhex
 from solaar import NAME, __version__
 
 _hidpp10 = hidpp10.Hidpp10()
+_hidpp20 = hidpp20.Hidpp20()
 
 
 def _print_receiver(receiver):
@@ -233,7 +233,7 @@ def _print_device(dev, num=None):
                 else:
                     mode = "On-Board"
                 print("            Device Mode: %s" % mode)
-            elif _hidpp20.battery_functions.get(feature, None):
+            elif hidpp20.battery_functions.get(feature, None):
                 print("", end="       ")
                 _battery_line(dev)
             for setting in dev_settings:
@@ -247,7 +247,7 @@ def _print_device(dev, num=None):
                         print("            %s (saved): %s" % (setting.label, v))
                     try:
                         v = setting.val_to_string(setting.read(False))
-                    except _hidpp20.FeatureCallError as e:
+                    except exceptions.FeatureCallError as e:
                         v = "HID++ error " + str(e)
                     except AssertionError as e:
                         v = "AssertionError " + str(e)