Skip to content

Commit

Permalink
device: clean up device code
Browse files Browse the repository at this point in the history
  • Loading branch information
@pfps
pfps committed Feb 22, 2024
1 parent 1d2c908 commit 74dcdc3
Showing 1 changed file with 42 additions and 48 deletions.
90 changes: 42 additions & 48 deletions lib/logitech_receiver/device.py
View file
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -86,25 +80,25 @@ 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)
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

Expand All @@ -117,9 +111,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)

Expand All @@ -141,7 +135,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:
Expand All @@ -161,11 +155,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:
Expand All @@ -192,16 +186,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
Expand All @@ -222,28 +216,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
Expand All @@ -252,21 +246,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
Expand Down Expand Up @@ -303,7 +297,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)
Expand All @@ -316,11 +310,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:
Expand All @@ -338,19 +332,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
Expand Down Expand Up @@ -389,7 +383,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,
Expand All @@ -401,14 +395,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
Expand All @@ -423,7 +417,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)
Expand All @@ -438,7 +432,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
Expand Down

0 comments on commit 74dcdc3

Please sign in to comment.