diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index d903707f17..9f123c3543 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -1,6 +1,5 @@ -# -*- python-mode -*- - ## Copyright (C) 2012-2013 Daniel Pavel +## Copyright (C) 2014-2024 Solaar Contributors https://pwr-solaar.github.io/Solaar/ ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by @@ -26,24 +25,19 @@ import hidapi as _hid import solaar.configuration as _configuration -from . import base as _base -from . import descriptors as _descriptors -from . import exceptions -from . import hidpp10 as _hidpp10 -from . import hidpp10_constants as _hidpp10_constants -from . import hidpp20 as _hidpp20 +from . import base, descriptors, exceptions, hidpp10, hidpp10_constants, hidpp20 from .settings_templates import check_feature_settings as _check_feature_settings logger = logging.getLogger(__name__) -_R = _hidpp10_constants.REGISTERS -_IR = _hidpp10_constants.INFO_SUBREGISTERS +_R = hidpp10_constants.REGISTERS +_IR = hidpp10_constants.INFO_SUBREGISTERS class Device: instances = [] - read_register = _hidpp10.read_register - write_register = _hidpp10.write_register + read_register = hidpp10.read_register + write_register = hidpp10.write_register def __init__(self, receiver, number, online, pairing_info=None, handle=None, device_info=None, setting_callback=None): assert receiver or device_info @@ -63,7 +57,7 @@ def __init__(self, receiver, number, online, pairing_info=None, handle=None, dev self.bluetooth = device_info.bus_id == 0x0005 if device_info else False # Bluetooth needs long messages self.setting_callback = setting_callback self.wpid = pairing_info["wpid"] if pairing_info else None # the Wireless PID is unique per device model - self._kind = pairing_info["kind"] if pairing_info else None # mouse, keyboard, etc (see _hidpp10.DEVICE_KIND) + self._kind = pairing_info["kind"] if pairing_info else None # mouse, keyboard, etc (see hidpp10.DEVICE_KIND) self._serial = pairing_info["serial"] if pairing_info else None # serial number (an 8-char hex string) self._polling_rate = pairing_info["polling"] if pairing_info else None self._power_switch = pairing_info["power_switch"] if pairing_info else None @@ -86,25 +80,29 @@ def __init__(self, receiver, number, online, pairing_info=None, handle=None, dev self.path = _hid.find_paired_node(receiver.path, number, 1) if receiver else None if not self.handle: try: - self.handle = _base.open_path(self.path) if self.path else None + self.handle = base.open_path(self.path) if self.path else None except Exception: # maybe the device wasn't set up try: time.sleep(1) - self.handle = _base.open_path(self.path) if self.path else None + self.handle = base.open_path(self.path) if self.path else None except Exception: # give up self.handle = None # should this give up completely? if receiver: if not self.wpid: - raise exceptions.NoSuchDevice(number=number, receiver=receiver, error="no wpid for device connected to receiver") - self.descriptor = _descriptors.get_wpid(self.wpid) + raise exceptions.NoSuchDevice( + number=number, receiver=receiver, error="no wpid for device connected to receiver" + ) + self.descriptor = descriptors.get_wpid(self.wpid) if self.descriptor is None: codename = self.receiver.device_codename(self.number) # Last chance to get a descriptor, may fail if codename: self._codename = codename - self.descriptor = _descriptors.get_codename(self._codename) + self.descriptor = descriptors.get_codename(self._codename) else: - self.descriptor = _descriptors.get_btid(self.product_id) if self.bluetooth else _descriptors.get_usbid(self.product_id) + self.descriptor = ( + descriptors.get_btid(self.product_id) if self.bluetooth else descriptors.get_usbid(self.product_id) + ) if self.number is None: # for direct-connected devices get 'number' from descriptor protocol else use 0xFF self.number = 0x00 if self.descriptor and self.descriptor.protocol and self.descriptor.protocol < 2.0 else 0xFF @@ -117,9 +115,9 @@ def __init__(self, receiver, number, online, pairing_info=None, handle=None, dev self._protocol = self.descriptor.protocol if self.descriptor.protocol else None if self._protocol is not None: - self.features = None if self._protocol < 2.0 else _hidpp20.FeaturesArray(self) + self.features = None if self._protocol < 2.0 else hidpp20.FeaturesArray(self) else: - self.features = _hidpp20.FeaturesArray(self) # may be a 2.0 device; if not, it will fix itself later + self.features = hidpp20.FeaturesArray(self) # may be a 2.0 device; if not, it will fix itself later Device.instances.append(self) @@ -141,7 +139,7 @@ def codename(self): 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: @@ -161,11 +159,11 @@ def name(self): 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: @@ -192,16 +190,16 @@ def tid_map(self): @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) + self._firmware = hidpp10.get_firmware(self) return self._firmware or () @property @@ -222,28 +220,28 @@ def power_switch_location(self): @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 @property def led_effects(self): if not self._led_effects and self.online and self.protocol >= 2.0: - self._led_effects = _hidpp20.LEDEffectsInfo(self) + self._led_effects = hidpp20.LEDEffectsInfo(self) return self._led_effects @property 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 @@ -252,21 +250,21 @@ def gestures(self): 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 @@ -303,7 +301,7 @@ def settings(self): 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) @@ -316,11 +314,11 @@ def persister(self): def battery(self): # None or level, next, status, voltage if self.protocol < 2.0: - return _hidpp10.get_battery(self) + return hidpp10.get_battery(self) 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: @@ -338,19 +336,19 @@ def enable_connection_notifications(self, enable=True): if enable: set_flag_bits = ( - _hidpp10_constants.NOTIFICATION_FLAG.battery_status - | _hidpp10_constants.NOTIFICATION_FLAG.keyboard_illumination - | _hidpp10_constants.NOTIFICATION_FLAG.wireless - | _hidpp10_constants.NOTIFICATION_FLAG.software_present + hidpp10_constants.NOTIFICATION_FLAG.battery_status + | hidpp10_constants.NOTIFICATION_FLAG.keyboard_illumination + | hidpp10_constants.NOTIFICATION_FLAG.wireless + | hidpp10_constants.NOTIFICATION_FLAG.software_present ) else: set_flag_bits = 0 - ok = _hidpp10.set_notification_flags(self, set_flag_bits) + ok = hidpp10.set_notification_flags(self, set_flag_bits) if not ok: logger.warning("%s: failed to %s device notifications", self, "enable" if enable else "disable") - flag_bits = _hidpp10.get_notification_flags(self) - flag_names = None if flag_bits is None else tuple(_hidpp10_constants.NOTIFICATION_FLAG.flag_names(flag_bits)) + flag_bits = hidpp10.get_notification_flags(self) + flag_names = None if flag_bits is None else tuple(hidpp10_constants.NOTIFICATION_FLAG.flag_names(flag_bits)) if logger.isEnabledFor(logging.INFO): logger.info("%s: device notifications %s %s", self, "enabled" if enable else "disabled", flag_names) return flag_bits if ok else None @@ -389,7 +387,7 @@ def request(self, request_id, *params, no_reply=False): long = self.hidpp_long is True or ( self.hidpp_long is None and (self.bluetooth or self._protocol is not None and self._protocol >= 2.0) ) - return _base.request( + return base.request( self.handle or self.receiver.handle, self.number, request_id, @@ -401,14 +399,14 @@ def request(self, request_id, *params, no_reply=False): def feature_request(self, feature, function=0x00, *params, no_reply=False): if self.protocol >= 2.0: - return _hidpp20.feature_request(self, feature, function, *params, no_reply=no_reply) + return hidpp20.feature_request(self, feature, function, *params, no_reply=no_reply) def ping(self): """Checks if the device is online, returns True of False""" long = self.hidpp_long is True or ( self.hidpp_long is None and (self.bluetooth or self._protocol is not None and self._protocol >= 2.0) ) - protocol = _base.ping(self.handle or self.receiver.handle, self.number, long_message=long) + protocol = base.ping(self.handle or self.receiver.handle, self.number, long_message=long) self.online = protocol is not None if protocol: self._protocol = protocol @@ -423,7 +421,7 @@ def open(self, device_info, setting_callback=None): :returns: An open file handle for the found receiver, or None. """ try: - handle = _base.open_path(device_info.path) + handle = base.open_path(device_info.path) if handle: # a direct connected device might not be online (as reported by user) return Device(None, None, None, handle=handle, device_info=device_info, setting_callback=setting_callback) @@ -438,7 +436,7 @@ def close(self): handle, self.handle = self.handle, None if self in Device.instances: Device.instances.remove(self) - return handle and _base.close(handle) + return handle and base.close(handle) def __index__(self): return self.number diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index ba3f7c645e..df11a43dd8 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -1,6 +1,5 @@ -# -*- python-mode -*- - ## Copyright (C) 2012-2013 Daniel Pavel +## Copyright (C) 2014-2024 Solaar Contributors https://pwr-solaar.github.io/Solaar/ ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by @@ -16,8 +15,6 @@ ## with this program; if not, write to the Free Software Foundation, Inc., ## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. -# Logitech Unifying Receiver API. - import logging import socket import threading as _threading @@ -51,10 +48,6 @@ def hexint_presenter(dumper, data): _yaml.add_representer(int, hexint_presenter) -# -# -# - class FeaturesArray(dict): def __init__(self, device): @@ -278,7 +271,9 @@ def _getCidReporting(self): raise exceptions.FeatureCallError(msg="No reply from device.") except exceptions.FeatureCallError: # if the key hasn't ever been configured only produce a warning if logger.isEnabledFor(logging.WARNING): - logger.warn(f"Feature Call Error in _getCidReporting on device {self._device} for cid {self._cid} - use defaults") + logger.warn( + f"Feature Call Error in _getCidReporting on device {self._device} for cid {self._cid} - use defaults" + ) # Clear flags and set mapping target to self self._mapping_flags = 0 self._mapped_to = self._cid @@ -545,7 +540,9 @@ def _query_key(self, index: int): if keydata: key = _unpack("!H", keydata[:2])[0] try: - mapped_data = feature_request(self.device, FEATURE.PERSISTENT_REMAPPABLE_ACTION, 0x30, key & 0xFF00, key & 0xFF, 0xFF) + mapped_data = feature_request( + self.device, FEATURE.PERSISTENT_REMAPPABLE_ACTION, 0x30, key & 0xFF00, key & 0xFF, 0xFF + ) if mapped_data: _ignore, _ignore, actionId, remapped, modifiers, status = _unpack("!HBBHBB", mapped_data[:8]) except Exception: @@ -872,7 +869,9 @@ def __init__(self, device): if not response: raise exceptions.FeatureCallError(msg="No reply from device.") self.device = device - self.enabled, self.options, supported, effects, self.level, self.dho, self.dhi, self.dpow = _unpack(" 6: self.max_devices = product_info.get("max_devices", 1) @@ -107,7 +98,7 @@ def __del__(self): @property def firmware(self): if self._firmware is None and self.handle: - self._firmware = _hidpp10.get_firmware(self) + self._firmware = hidpp10.get_firmware(self) return self._firmware # how many pairings remain (None for unknown, -1 for unlimited) @@ -127,19 +118,19 @@ def enable_connection_notifications(self, enable=True): if enable: set_flag_bits = ( - _hidpp10_constants.NOTIFICATION_FLAG.battery_status - | _hidpp10_constants.NOTIFICATION_FLAG.wireless - | _hidpp10_constants.NOTIFICATION_FLAG.software_present + hidpp10_constants.NOTIFICATION_FLAG.battery_status + | hidpp10_constants.NOTIFICATION_FLAG.wireless + | hidpp10_constants.NOTIFICATION_FLAG.software_present ) else: set_flag_bits = 0 - ok = _hidpp10.set_notification_flags(self, set_flag_bits) + ok = hidpp10.set_notification_flags(self, set_flag_bits) if ok is None: logger.warning("%s: failed to %s receiver notifications", self, "enable" if enable else "disable") return None - flag_bits = _hidpp10.get_notification_flags(self) - flag_names = None if flag_bits is None else tuple(_hidpp10_constants.NOTIFICATION_FLAG.flag_names(flag_bits)) + flag_bits = hidpp10.get_notification_flags(self) + flag_names = None if flag_bits is None else tuple(hidpp10_constants.NOTIFICATION_FLAG.flag_names(flag_bits)) if logger.isEnabledFor(logging.INFO): logger.info("%s: receiver notifications %s => %s", self, "enabled" if enable else "disabled", flag_names) return flag_bits @@ -161,9 +152,9 @@ def device_pairing_information(self, n: int) -> dict: if self.receiver_kind == "bolt": pair_info = self.read_register(_R.receiver_info, _IR.bolt_pairing_information + n) if pair_info: - wpid = _strhex(pair_info[3:4]) + _strhex(pair_info[2:3]) - kind = _hidpp10_constants.DEVICE_KIND[pair_info[1] & 0x0F] - serial = _strhex(pair_info[4:8]) + wpid = (pair_info[3:4] + pair_info[2:3]).hex().upper() + kind = hidpp10_constants.DEVICE_KIND[pair_info[1] & 0x0F] + serial = pair_info[4:8].hex().upper() return {"wpid": wpid, "kind": kind, "polling": None, "serial": serial, "power_switch": "(unknown)"} else: raise exceptions.NoSuchDevice(number=n, receiver=self, error="can't read Bolt pairing register") @@ -172,31 +163,31 @@ def device_pairing_information(self, n: int) -> dict: power_switch = "(unknown)" pair_info = self.read_register(_R.receiver_info, _IR.pairing_information + n - 1) if pair_info: # either a Unifying receiver or a Unifying-ready receiver - wpid = _strhex(pair_info[3:5]) - kind = _hidpp10_constants.DEVICE_KIND[pair_info[7] & 0x0F] + wpid = pair_info[3:5].hex().upper() + kind = hidpp10_constants.DEVICE_KIND[pair_info[7] & 0x0F] polling_rate = str(ord(pair_info[2:3])) + "ms" elif self.receiver_kind == "27Mz": # 27Mhz receiver, extract WPID from udev path wpid = _hid.find_paired_node_wpid(self.path, n) if not wpid: logger.error("Unable to get wpid from udev for device %d of %s", n, self) raise exceptions.NoSuchDevice(number=n, receiver=self, error="Not present 27Mhz device") - kind = _hidpp10_constants.DEVICE_KIND[self.get_kind_from_index(n)] + kind = hidpp10_constants.DEVICE_KIND[self.get_kind_from_index(n)] elif not self.receiver_kind == "unifying": # may be an old Nano receiver device_info = self.read_register(_R.receiver_info, 0x04) if device_info: - wpid = _strhex(device_info[3:5]) - kind = _hidpp10_constants.DEVICE_KIND[0x00] # unknown kind + wpid = device_info[3:5].hex().upper() + kind = hidpp10_constants.DEVICE_KIND[0x00] # unknown kind else: raise exceptions.NoSuchDevice(number=n, receiver=self, error="read pairing information - non-unifying") else: raise exceptions.NoSuchDevice(number=n, receiver=self, error="read pairing information") pair_info = self.read_register(_R.receiver_info, _IR.extended_pairing_information + n - 1) if pair_info: - power_switch = _hidpp10_constants.POWER_SWITCH_LOCATION[pair_info[9] & 0x0F] + power_switch = hidpp10_constants.POWER_SWITCH_LOCATION[pair_info[9] & 0x0F] else: # some Nano receivers? pair_info = self.read_register(0x2D5) if pair_info: - serial = _strhex(pair_info[1:5]) + serial = pair_info[1:5].hex().upper() return {"wpid": wpid, "kind": kind, "polling": polling_rate, "serial": serial, "power_switch": power_switch} def get_kind_from_index(self, index): @@ -233,12 +224,12 @@ def register_new_device(self, number, notification=None): if notification is not None: online = not bool(ord(notification.data[0:1]) & 0x40) # the rest may be redundant, but keep it around for now - info["wpid"] = _strhex(notification.data[2:3] + notification.data[1:2]) + info["wpid"] = (notification.data[2:3] + notification.data[1:2]).hex().upper() kind = ord(notification.data[0:1]) & 0x0F if self.receiver_kind == "27Mhz": # get 27Mhz wpid and set kind based on index - info["wpid"] = "00" + _strhex(notification.data[2:3]) + info["wpid"] = "00" + notification.data[2:3].hex().upper() kind = self.get_kind_from_index(number) - info["kind"] = _hidpp10_constants.DEVICE_KIND[kind] + info["kind"] = hidpp10_constants.DEVICE_KIND[kind] else: online = True dev = Device(self, number, online, pairing_info=info, setting_callback=self.setting_callback) @@ -289,8 +280,8 @@ def request(self, request_id, *params): if bool(self): return _base.request(self.handle, 0xFF, request_id, *params) - read_register = _hidpp10.read_register - write_register = _hidpp10.write_register + read_register = hidpp10.read_register + write_register = hidpp10.write_register def __iter__(self): connected_devices = self.count() diff --git a/pyproject.toml b/pyproject.toml index c4fa1ea361..285ad5cac5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,5 +1,5 @@ [tool.ruff] -line-length = 140 +line-length = 127 target-version = "py37" [tool.ruff.lint]