RomanNum3ral
/
Solaar
mirror of https://github.com/pwr-Solaar/Solaar.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

flake8: initial fix 

Signed-off-by: Filipe Laíns <lains@archlinux.org>
This commit is contained in:
Filipe Laíns 2020-07-02 14:35:04 +01:00 committed by Filipe Laíns
parent de5514aa23
commit 627185079f
35 changed files with 397 additions and 401 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
bin/solaar
View File

@ -36,9 +36,8 @@ def init_paths():
sys.path[0].encode(sys.getfilesystemencoding())
except UnicodeError:
sys.stderr.write(
'ERROR: Solaar cannot recognize encoding of filesystem path, this may happen because non UTF-8 characters in the pathname.\n'
)
sys.stderr.write('ERROR: Solaar cannot recognize encoding of filesystem path, '
'this may happen because non UTF-8 characters in the pathname.\n')
sys.exit(1)
prefix = _path.normpath(_path.join(_path.realpath(decoded_path), '..'))

11
lib/hidapi/__init__.py
View File

@ -20,6 +20,15 @@
from __future__ import absolute_import, division, print_function, unicode_literals
from hidapi.udev import close, enumerate, get_manufacturer, get_product, get_serial, monitor_glib, open, open_path, read, write
from hidapi.udev import close # noqa: F401
from hidapi.udev import enumerate # noqa: F401
from hidapi.udev import get_manufacturer # noqa: F401
from hidapi.udev import get_product # noqa: F401
from hidapi.udev import get_serial # noqa: F401
from hidapi.udev import monitor_glib # noqa: F401
from hidapi.udev import open # noqa: F401
from hidapi.udev import open_path # noqa: F401
from hidapi.udev import read # noqa: F401
from hidapi.udev import write # noqa: F401
__version__ = '0.9'

14
lib/hidapi/hidconsole.py
View File

@ -45,15 +45,15 @@ start_time = time.time()
strhex = lambda d: hexlify(d).decode('ascii').upper()
try:
unicode
unicode # noqa: F821
# this is certanly Python 2
is_string = lambda d: isinstance(d, unicode)
is_string = lambda d: isinstance(d, unicode) # noqa: F821
# no easy way to distinguish between b'' and '' :(
# or (isinstance(d, str) \
# and not any((chr(k) in d for k in range(0x00, 0x1F))) \
# and not any((chr(k) in d for k in range(0x80, 0xFF))) \
# )
except:
# and not any((chr(k) in d for k in range(0x00, 0x1F))) \
# and not any((chr(k) in d for k in range(0x80, 0xFF))) \
# )
except Exception:
# this is certanly Python 3
# In Py3, unicode and str are equal (the unicode object does not exist)
is_string = lambda d: isinstance(d, str)
@ -225,7 +225,7 @@ def main():
'.hidconsole-history')
try:
readline.read_history_file(args.history)
except:
except Exception:
# file may not exist yet
pass

130
lib/hidapi/udev.py
View File

@ -63,19 +63,19 @@ del namedtuple
def init():
"""This function is a no-op, and exists only to match the native hidapi
implementation.
implementation.
:returns: ``True``.
"""
:returns: ``True``.
"""
return True
def exit():
"""This function is a no-op, and exists only to match the native hidapi
implementation.
implementation.
:returns: ``True``.
"""
:returns: ``True``.
"""
return True
@ -157,12 +157,12 @@ def monitor_glib(callback, *device_filters):
# already existing devices
# for device in c.list_devices(subsystem='hidraw'):
# # print (device, dict(device), dict(device.attributes))
# for filter in device_filters:
# d_info = _match('add', device, *filter)
# if d_info:
# GLib.idle_add(callback, 'add', d_info)
# break
# # print (device, dict(device), dict(device.attributes))
# for filter in device_filters:
# d_info = _match('add', device, *filter)
# if d_info:
# GLib.idle_add(callback, 'add', d_info)
# break
m = _Monitor.from_netlink(c)
m.filter_by(subsystem='hidraw')
@ -195,7 +195,7 @@ def monitor_glib(callback, *device_filters):
GLib.io_add_watch(m, GLib.PRIORITY_LOW, GLib.IO_IN,
_process_udev_event, callback, device_filters)
# print ("did io_add_watch with priority")
except:
except Exception:
GLib.io_add_watch(m, GLib.IO_IN, _process_udev_event, callback,
device_filters)
# print ("did io_add_watch")
@ -206,11 +206,11 @@ def monitor_glib(callback, *device_filters):
def enumerate(usb_id):
"""Enumerate the HID Devices.
List all the HID devices attached to the system, optionally filtering by
vendor_id, product_id, and/or interface_number.
List all the HID devices attached to the system, optionally filtering by
vendor_id, product_id, and/or interface_number.
:returns: a list of matching ``DeviceInfo`` tuples.
"""
:returns: a list of matching ``DeviceInfo`` tuples.
"""
for dev in _Context().list_devices(subsystem='hidraw'):
dev_info = _match('add', dev, usb_id)
@ -221,10 +221,10 @@ def enumerate(usb_id):
def open(vendor_id, product_id, serial=None):
"""Open a HID device by its Vendor ID, Product ID and optional serial number.
If no serial is provided, the first device with the specified IDs is opened.
If no serial is provided, the first device with the specified IDs is opened.
:returns: an opaque device handle, or ``None``.
"""
:returns: an opaque device handle, or ``None``.
"""
for device in enumerate(vendor_id, product_id):
if serial is None or serial == device.serial:
return open_path(device.path)
@ -233,11 +233,11 @@ def open(vendor_id, product_id, serial=None):
def open_path(device_path):
"""Open a HID device by its path name.
:param device_path: the path of a ``DeviceInfo`` tuple returned by
enumerate().
:param device_path: the path of a ``DeviceInfo`` tuple returned by
enumerate().
:returns: an opaque device handle, or ``None``.
"""
:returns: an opaque device handle, or ``None``.
"""
assert device_path
assert device_path.startswith('/dev/hidraw')
return _os.open(device_path, _os.O_RDWR | _os.O_SYNC)
@ -246,8 +246,8 @@ def open_path(device_path):
def close(device_handle):
"""Close a HID device.
:param device_handle: a device handle returned by open() or open_path().
"""
:param device_handle: a device handle returned by open() or open_path().
"""
assert device_handle
_os.close(device_handle)
@ -255,24 +255,24 @@ def close(device_handle):
def write(device_handle, data):
"""Write an Output report to a HID device.
:param device_handle: a device handle returned by open() or open_path().
:param data: the data bytes to send including the report number as the
first byte.
:param device_handle: a device handle returned by open() or open_path().
:param data: the data bytes to send including the report number as the
first byte.
The first byte of data[] must contain the Report ID. For
devices which only support a single report, this must be set
to 0x0. The remaining bytes contain the report data. Since
the Report ID is mandatory, calls to hid_write() will always
contain one more byte than the report contains. For example,
if a hid report is 16 bytes long, 17 bytes must be passed to
hid_write(), the Report ID (or 0x0, for devices with a
single report), followed by the report data (16 bytes). In
this example, the length passed in would be 17.
The first byte of data[] must contain the Report ID. For
devices which only support a single report, this must be set
to 0x0. The remaining bytes contain the report data. Since
the Report ID is mandatory, calls to hid_write() will always
contain one more byte than the report contains. For example,
if a hid report is 16 bytes long, 17 bytes must be passed to
hid_write(), the Report ID (or 0x0, for devices with a
single report), followed by the report data (16 bytes). In
this example, the length passed in would be 17.
write() will send the data on the first OUT endpoint, if
one exists. If it does not, it will send the data through
the Control Endpoint (Endpoint 0).
"""
write() will send the data on the first OUT endpoint, if
one exists. If it does not, it will send the data through
the Control Endpoint (Endpoint 0).
"""
assert device_handle
assert data
assert isinstance(data, bytes), (repr(data), type(data))
@ -296,19 +296,19 @@ def write(device_handle, data):
def read(device_handle, bytes_count, timeout_ms=-1):
"""Read an Input report from a HID device.
:param device_handle: a device handle returned by open() or open_path().
:param bytes_count: maximum number of bytes to read.
:param timeout_ms: can be -1 (default) to wait for data indefinitely, 0 to
read whatever is in the device's input buffer, or a positive integer to
wait that many milliseconds.
:param device_handle: a device handle returned by open() or open_path().
:param bytes_count: maximum number of bytes to read.
:param timeout_ms: can be -1 (default) to wait for data indefinitely, 0 to
read whatever is in the device's input buffer, or a positive integer to
wait that many milliseconds.
Input reports are returned to the host through the INTERRUPT IN endpoint.
The first byte will contain the Report number if the device uses numbered
reports.
Input reports are returned to the host through the INTERRUPT IN endpoint.
The first byte will contain the Report number if the device uses numbered
reports.
:returns: the data packet read, an empty bytes string if a timeout was
reached, or None if there was an error while reading.
"""
:returns: the data packet read, an empty bytes string if a timeout was
reached, or None if there was an error while reading.
"""
assert device_handle
timeout = None if timeout_ms < 0 else timeout_ms / 1000.0
rlist, wlist, xlist = _select([device_handle], [], [device_handle],
@ -339,24 +339,24 @@ _DEVICE_STRINGS = {
def get_manufacturer(device_handle):
"""Get the Manufacturer String from a HID device.
:param device_handle: a device handle returned by open() or open_path().
"""
:param device_handle: a device handle returned by open() or open_path().
"""
return get_indexed_string(device_handle, 0)
def get_product(device_handle):
"""Get the Product String from a HID device.
:param device_handle: a device handle returned by open() or open_path().
"""
:param device_handle: a device handle returned by open() or open_path().
"""
return get_indexed_string(device_handle, 1)
def get_serial(device_handle):
"""Get the serial number from a HID device.
:param device_handle: a device handle returned by open() or open_path().
"""
:param device_handle: a device handle returned by open() or open_path().
"""
serial = get_indexed_string(device_handle, 2)
if serial is not None:
return ''.join(hex(ord(c)) for c in serial)
@ -365,13 +365,13 @@ def get_serial(device_handle):
def get_indexed_string(device_handle, index):
"""Get a string from a HID device, based on its string index.
Note: currently not working in the ``hidraw`` native implementation.
Note: currently not working in the ``hidraw`` native implementation.
:param device_handle: a device handle returned by open() or open_path().
:param index: the index of the string to get.
:returns: the value corresponding to index, or None if no value found
:rtype: bytes or NoneType
"""
:param device_handle: a device handle returned by open() or open_path().
:param index: the index of the string to get.
:returns: the value corresponding to index, or None if no value found
:rtype: bytes or NoneType
"""
try:
key = _DEVICE_STRINGS[index]
except KeyError:

14
lib/logitech_receiver/__init__.py
View File

@ -33,18 +33,18 @@ from __future__ import absolute_import, division, print_function, unicode_litera
import logging
from . import listener, status
from .base import DeviceUnreachable, NoReceiver, NoSuchDevice
from .common import strhex
from .hidpp20 import FeatureCallError, FeatureNotSupported
from .receiver import PairedDevice, Receiver
from . import listener, status # noqa: F401
from .base import DeviceUnreachable, NoReceiver, NoSuchDevice # noqa: F401
from .common import strhex # noqa: F401
from .hidpp20 import FeatureCallError, FeatureNotSupported # noqa: F401
from .receiver import PairedDevice, Receiver # noqa: F401
_DEBUG = logging.DEBUG
_log = logging.getLogger(__name__)
_log.setLevel(logging.root.level)
# if logging.root.level > logging.DEBUG:
# _log.addHandler(logging.NullHandler())
# _log.propagate = 0
# _log.addHandler(logging.NullHandler())
# _log.propagate = 0
del logging

118
lib/logitech_receiver/base.py
View File

@ -72,9 +72,9 @@ _PING_TIMEOUT = DEFAULT_TIMEOUT * 2
class NoReceiver(_KwException):
"""Raised when trying to talk through a previously open handle, when the
receiver is no longer available. Should only happen if the receiver is
physically disconnected from the machine, or its kernel driver module is
unloaded."""
receiver is no longer available. Should only happen if the receiver is
physically disconnected from the machine, or its kernel driver module is
unloaded."""
pass
@ -113,24 +113,24 @@ def notify_on_receivers_glib(callback):
def open_path(path):
"""Checks if the given Linux device path points to the right UR device.
:param path: the Linux device path.
:param path: the Linux device path.
The UR physical device may expose multiple linux devices with the same
interface, so we have to check for the right one. At this moment the only
way to distinguish betheen them is to do a test ping on an invalid
(attached) device number (i.e., 0), expecting a 'ping failed' reply.
The UR physical device may expose multiple linux devices with the same
interface, so we have to check for the right one. At this moment the only
way to distinguish betheen them is to do a test ping on an invalid
(attached) device number (i.e., 0), expecting a 'ping failed' reply.
:returns: an open receiver handle if this is the right Linux device, or
``None``.
"""
:returns: an open receiver handle if this is the right Linux device, or
``None``.
"""
return _hid.open_path(path)
def open():
"""Opens the first Logitech Unifying Receiver found attached to the machine.
:returns: An open file handle for the found receiver, or ``None``.
"""
:returns: An open file handle for the found receiver, or ``None``.
"""
for rawdevice in receivers():
handle = open_path(rawdevice.path)
if handle:
@ -147,7 +147,7 @@ def close(handle):
handle.close()
# _log.info("closed receiver handle %r", handle)
return True
except:
except Exception:
# _log.exception("closing receiver handle %r", handle)
pass
@ -157,16 +157,16 @@ def close(handle):
def write(handle, devnumber, data):
"""Writes some data to the receiver, addressed to a certain device.
:param handle: an open UR handle.
:param devnumber: attached device number.
:param data: data to send, up to 5 bytes.
:param handle: an open UR handle.
:param devnumber: attached device number.
:param data: data to send, up to 5 bytes.
The first two (required) bytes of data must be the SubId and address.
The first two (required) bytes of data must be the SubId and address.
:raises NoReceiver: if the receiver is no longer available, i.e. has
been physically removed from the machine, or the kernel driver has been
unloaded. The handle will be closed automatically.
"""
:raises NoReceiver: if the receiver is no longer available, i.e. has
been physically removed from the machine, or the kernel driver has been
unloaded. The handle will be closed automatically.
"""
# the data is padded to either 5 or 18 bytes
assert data is not None
assert isinstance(data, bytes), (repr(data), type(data))
@ -190,17 +190,17 @@ def write(handle, devnumber, data):
def read(handle, timeout=DEFAULT_TIMEOUT):
"""Read some data from the receiver. Usually called after a write (feature
call), to get the reply.
call), to get the reply.
:param: handle open handle to the receiver
:param: timeout how long to wait for a reply, in seconds
:param: handle open handle to the receiver
:param: timeout how long to wait for a reply, in seconds
:returns: a tuple of (devnumber, message data), or `None`
:returns: a tuple of (devnumber, message data), or `None`
:raises NoReceiver: if the receiver is no longer available, i.e. has
been physically removed from the machine, or the kernel driver has been
unloaded. The handle will be closed automatically.
"""
:raises NoReceiver: if the receiver is no longer available, i.e. has
been physically removed from the machine, or the kernel driver has been
unloaded. The handle will be closed automatically.
"""
reply = _read(handle, timeout)
if reply:
return reply[1:]
@ -222,12 +222,12 @@ def check_message(data):
def _read(handle, timeout):
"""Read an incoming packet from the receiver.
:returns: a tuple of (report_id, devnumber, data), or `None`.
:returns: a tuple of (report_id, devnumber, data), or `None`.
:raises NoReceiver: if the receiver is no longer available, i.e. has
been physically removed from the machine, or the kernel driver has been
unloaded. The handle will be closed automatically.
"""
:raises NoReceiver: if the receiver is no longer available, i.e. has
been physically removed from the machine, or the kernel driver has been
unloaded. The handle will be closed automatically.
"""
try:
# convert timeout to milliseconds, the hidapi expects it
timeout = int(timeout * 1000)
@ -257,8 +257,8 @@ def _read(handle, timeout):
def _skip_incoming(handle, ihandle, notifications_hook):
"""Read anything already in the input buffer.
Used by request() and ping() before their write.
"""
Used by request() and ping() before their write.
"""
while True:
try:
@ -272,7 +272,7 @@ def _skip_incoming(handle, ihandle, notifications_hook):
if data:
if check_message(data): # only process messages that pass check
report_id = ord(data[:1])
# report_id = ord(data[:1])
if notifications_hook:
n = make_notification(ord(data[1:2]), data[2:])
if n:
@ -284,7 +284,7 @@ def _skip_incoming(handle, ihandle, notifications_hook):
def make_notification(devnumber, data):
"""Guess if this is a notification (and not just a request reply), and
return a Notification tuple if it is."""
return a Notification tuple if it is."""
sub_id = ord(data[:1])
if sub_id & 0x80 == 0x80:
@ -299,13 +299,13 @@ def make_notification(devnumber, data):
address = ord(data[1:2])
if (
# standard HID++ 1.0 notification, SubId may be 0x40 - 0x7F
(sub_id >= 0x40) or
(sub_id >= 0x40) or # noqa: E131
# 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
# custom HID++1.0 illumination event, where SubId is 0x17
(sub_id == 0x17 and len(data) == 5) or
# HID++ 2.0 feature notifications have the SoftwareID 0
(address & 0x0F == 0x00)):
(address & 0x0F == 0x00)): # noqa: E129
return _HIDPP_Notification(devnumber, sub_id, address, data[2:])
@ -325,14 +325,14 @@ del namedtuple
def request(handle, devnumber, request_id, *params):
"""Makes a feature call to a device and waits for a matching reply.
This function will wait for a matching reply indefinitely.
This function will wait for a matching reply indefinitely.
:param handle: an open UR handle.
:param devnumber: attached device number.
:param request_id: a 16-bit integer.
:param params: parameters for the feature call, 3 to 16 bytes.
:returns: the reply data, or ``None`` if some error occurred.
"""
:param handle: an open UR handle.
:param devnumber: attached device number.
:param request_id: a 16-bit integer.
:param params: parameters for the feature call, 3 to 16 bytes.
:returns: the reply data, or ``None`` if some error occurred.
"""
# import inspect as _inspect
# print ('\n '.join(str(s) for s in _inspect.stack()))
@ -357,7 +357,7 @@ def request(handle, devnumber, request_id, *params):
else:
params = b''
# if _log.isEnabledFor(_DEBUG):
# _log.debug("(%s) device %d request_id {%04X} params [%s]", handle, devnumber, request_id, _strhex(params))
# _log.debug("(%s) device %d request_id {%04X} params [%s]", handle, devnumber, request_id, _strhex(params))
request_data = _pack('!H', request_id) + params
ihandle = int(handle)
@ -380,12 +380,12 @@ def request(handle, devnumber, request_id, *params):
error = ord(reply_data[3:4])
# if error == _hidpp10.ERROR.resource_error: # device unreachable
# _log.warn("(%s) device %d error on request {%04X}: unknown device", handle, devnumber, request_id)
# raise DeviceUnreachable(number=devnumber, request=request_id)
# _log.warn("(%s) device %d error on request {%04X}: unknown device", handle, devnumber, request_id)
# raise DeviceUnreachable(number=devnumber, request=request_id)
# if error == _hidpp10.ERROR.unknown_device: # unknown device
# _log.error("(%s) device %d error on request {%04X}: unknown device", handle, devnumber, request_id)
# raise NoSuchDevice(number=devnumber, request=request_id)
# _log.error("(%s) device %d error on request {%04X}: unknown device", handle, devnumber, request_id)
# raise NoSuchDevice(number=devnumber, request=request_id)
if _log.isEnabledFor(_DEBUG):
_log.debug(
@ -437,13 +437,13 @@ def request(handle, devnumber, request_id, *params):
if n:
notifications_hook(n)
# elif _log.isEnabledFor(_DEBUG):
# _log.debug("(%s) ignoring reply %02X [%s]", handle, reply_devnumber, _strhex(reply_data))
# _log.debug("(%s) ignoring reply %02X [%s]", handle, reply_devnumber, _strhex(reply_data))
# elif _log.isEnabledFor(_DEBUG):
# _log.debug("(%s) ignoring reply %02X [%s]", handle, reply_devnumber, _strhex(reply_data))
# _log.debug("(%s) ignoring reply %02X [%s]", handle, reply_devnumber, _strhex(reply_data))
delta = _timestamp() - request_started
# if _log.isEnabledFor(_DEBUG):
# _log.debug("(%s) still waiting for reply, delta %f", handle, delta)
# _log.debug("(%s) still waiting for reply, delta %f", handle, delta)
_log.warn('timeout (%0.2f/%0.2f) on device %d request {%04X} params [%s]',
delta, timeout, devnumber, request_id, _strhex(params))
@ -453,8 +453,8 @@ def request(handle, devnumber, request_id, *params):
def ping(handle, devnumber):
"""Check if a device is connected to the receiver.
:returns: The HID protocol supported by the device, as a floating point number, if the device is active.
"""
:returns: The HID protocol supported by the device, as a floating point number, if the device is active.
"""
if _log.isEnabledFor(_DEBUG):
_log.debug('(%s) pinging device %d', handle, devnumber)
@ -514,7 +514,7 @@ def ping(handle, devnumber):
if n:
notifications_hook(n)
# elif _log.isEnabledFor(_DEBUG):
# _log.debug("(%s) ignoring reply %02X [%s]", handle, reply_devnumber, _strhex(reply_data))
# _log.debug("(%s) ignoring reply %02X [%s]", handle, reply_devnumber, _strhex(reply_data))
delta = _timestamp() - request_started

16
lib/logitech_receiver/base_usb.py
View File

@ -22,19 +22,19 @@
from __future__ import absolute_import, division, print_function, unicode_literals
_DRIVER = ('hid-generic', 'generic-usb', 'logitech-djreceiver')
# max_devices is only used for receivers that do not support reading from _R.receiver_info offset 0x03, default to 1
# may_unpair is only used for receivers that do not support reading from _R.receiver_info offset 0x03, default to False
## should this last be changed so that may_unpair is used for all receivers? writing to _R.receiver_pairing doesn't seem right
# re_pairs determines whether a receiver pairs by replacing existing pairings, default to False
## currently only one receiver is so marked - should there be more?
_DRIVER = ('hid-generic', 'generic-usb', 'logitech-djreceiver')
_unifying_receiver = lambda product_id: {
'vendor_id': 0x046d,
'product_id': product_id,
'usb_interface': 2,
'hid_driver': _DRIVER,
'hid_driver': _DRIVER, # noqa: F821
'name': 'Unifying Receiver'
}
@ -42,7 +42,7 @@ _nano_receiver = lambda product_id: {
'vendor_id': 0x046d,
'product_id': product_id,
'usb_interface': 1,
'hid_driver': _DRIVER,
'hid_driver': _DRIVER, # noqa: F821
'name': 'Nano Receiver',
'may_unpair': False,
're_pairs': True
@ -52,7 +52,7 @@ _nano_receiver_max2 = lambda product_id: {
'vendor_id': 0x046d,
'product_id': product_id,
'usb_interface': 1,
'hid_driver': _DRIVER,
'hid_driver': _DRIVER, # noqa: F821
'name': 'Nano Receiver',
'max_devices': 2,
'may_unpair': False,
@ -63,7 +63,7 @@ _nano_receiver_maxn = lambda product_id, max: {
'vendor_id': 0x046d,
'product_id': product_id,
'usb_interface': 1,
'hid_driver': _DRIVER,
'hid_driver': _DRIVER, # noqa: F821
'name': 'Nano Receiver',
'max_devices': max,
'may_unpair': False,
@ -74,7 +74,7 @@ _lenovo_receiver = lambda product_id: {
'vendor_id': 0x17ef,
'product_id': product_id,
'usb_interface': 1,
'hid_driver': _DRIVER,
'hid_driver': _DRIVER, # noqa: F821
'name': 'Nano Receiver'
}
@ -82,7 +82,7 @@ _lightspeed_receiver = lambda product_id: {
'vendor_id': 0x046d,
'product_id': product_id,
'usb_interface': 2,
'hid_driver': _DRIVER,
'hid_driver': _DRIVER, # noqa: F821
'name': 'Lightspeed Receiver'
}

49
lib/logitech_receiver/common.py
View File

@ -26,20 +26,20 @@ from collections import namedtuple
from struct import pack, unpack
try:
unicode
unicode # noqa: F821
# if Python2, unicode_literals will mess our first (un)pack() argument
_pack_str = pack
_unpack_str = unpack
pack = lambda x, *args: _pack_str(str(x), *args)
unpack = lambda x, *args: _unpack_str(str(x), *args)
is_string = lambda d: isinstance(d, unicode) or isinstance(d, str)
is_string = lambda d: isinstance(d, unicode) or isinstance(d, str) # noqa: F821
# no easy way to distinguish between b'' and '' :(
# or (isinstance(d, str) \
# and not any((chr(k) in d for k in range(0x00, 0x1F))) \
# and not any((chr(k) in d for k in range(0x80, 0xFF))) \
# )
except:
# and not any((chr(k) in d for k in range(0x00, 0x1F))) \
# and not any((chr(k) in d for k in range(0x80, 0xFF))) \
# )
except Exception:
# this is certanly Python 3
# In Py3, unicode and str are equal (the unicode object does not exist)
is_string = lambda d: isinstance(d, str)
@ -52,8 +52,8 @@ except:
class NamedInt(int):
"""An reqular Python integer with an attached name.
Caution: comparison with strings will also match this NamedInt's name
(case-insensitive)."""
Caution: comparison with strings will also match this NamedInt's name
(case-insensitive)."""
def __new__(cls, value, name):
assert is_string(name)
obj = int.__new__(cls, value)
@ -92,16 +92,16 @@ class NamedInt(int):
class NamedInts(object):
"""An ordered set of NamedInt values.
Indexing can be made by int or string, and will return the corresponding
NamedInt if it exists in this set, or `None`.
Indexing can be made by int or string, and will return the corresponding
NamedInt if it exists in this set, or `None`.
Extracting slices will return all present NamedInts in the given interval
(extended slices are not supported).
Extracting slices will return all present NamedInts in the given interval
(extended slices are not supported).
Assigning a string to an indexed int will create a new NamedInt in this set;
if the value already exists in the set (int or string), ValueError will be
raised.
"""
Assigning a string to an indexed int will create a new NamedInt in this set;
if the value already exists in the set (int or string), ValueError will be
raised.
"""
__slots__ = ('__dict__', '_values', '_indexed', '_fallback')
def __init__(self, **kwargs):
@ -119,7 +119,8 @@ class NamedInts(object):
self.__dict__ = values
self._values = sorted(list(values.values()))
self._indexed = {int(v): v for v in self._values}
# assert len(values) == len(self._indexed), "(%d) %r\n=> (%d) %r" % (len(values), values, len(self._indexed), self._indexed)
# assert len(values) == len(self._indexed)
# "(%d) %r\n=> (%d) %r" % (len(values), values, len(self._indexed), self._indexed)
self._fallback = None
@classmethod
@ -237,8 +238,8 @@ def strhex(x):
def bytes2int(x):
"""Convert a bytes string to an int.
The bytes are assumed to be in most-significant-first order.
"""
The bytes are assumed to be in most-significant-first order.
"""
assert isinstance(x, bytes)
assert len(x) < 9
qx = (b'\x00' * 8) + x
@ -249,9 +250,9 @@ def bytes2int(x):
def int2bytes(x, count=None):
"""Convert an int to a bytes representation.
The bytes are ordered in most-significant-first order.
If 'count' is not given, the necessary number of bytes is computed.
"""
The bytes are ordered in most-significant-first order.
If 'count' is not given, the necessary number of bytes is computed.
"""
assert isinstance(x, int)
result = pack('!Q', x)
assert isinstance(result, bytes)
@ -268,8 +269,8 @@ def int2bytes(x, count=None):
class KwException(Exception):
"""An exception that remembers all arguments passed to the constructor.
They can be later accessed by simple member access.
"""
They can be later accessed by simple member access.
"""
def __init__(self, **kwargs):
super(KwException, self).__init__(kwargs)

42
lib/logitech_receiver/hidpp20.py
View File

@ -41,7 +41,8 @@ del getLogger
#
# <FeaturesSupported.xml sed '/LD_FID_/{s/.*LD_FID_/\t/;s/"[ \t]*Id="/=/;s/" \/>/,/p}' | sort -t= -k2
# additional features names taken from https://github.com/cvuchener/hidpp and https://github.com/Logitech/cpg-docs/tree/master/hidpp20
# additional features names taken from https://github.com/cvuchener/hidpp and
# https://github.com/Logitech/cpg-docs/tree/master/hidpp20
"""Possible features available on a Logitech device.
A particular device might not support all these features, and may support other
@ -414,8 +415,8 @@ class KeysArray(object):
remapped = key
except Exception:
remapped = key
remap_key = key
remap_flag = 0
# remap_key = key
# remap_flag = 0
remapped_text = special_keys.CONTROL[remapped]
self.keys[index] = _ReprogrammableKeyInfoV4(
@ -463,8 +464,8 @@ def feature_request(device, feature, function=0x00, *params):
def get_firmware(device):
"""Reads a device's firmware info.
:returns: a list of FirmwareInfo tuples, ordered by firmware layer.
"""
:returns: a list of FirmwareInfo tuples, ordered by firmware layer.
"""
count = feature_request(device, FEATURE.DEVICE_FW_VERSION)
if count:
count = ord(count[:1])
@ -493,31 +494,31 @@ def get_firmware(device):
fw.append(fw_info)
# if _log.isEnabledFor(_DEBUG):
# _log.debug("device %d firmware %s", devnumber, fw_info)
# _log.debug("device %d firmware %s", devnumber, fw_info)
return tuple(fw)
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.
"""
: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 _log.isEnabledFor(_DEBUG):
# _log.debug("device %d type %d = %s", devnumber, kind, DEVICE_KIND[kind])
# _log.debug("device %d type %d = %s", devnumber, kind, DEVICE_KIND[kind])
return 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.
"""
: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])
@ -582,10 +583,8 @@ def decipher_voltage(voltage_report):
charge_lvl = CHARGE_LEVEL.critical
if _log.isEnabledFor(_DEBUG):
_log.debug(
'device %d, battery voltage %d mV, charging = %s, charge status %d = %s, charge level %s, charge type %s',
device.number, voltage, status, (flags & 0x03), charge_sts,
charge_lvl, charge_type)
_log.debug('device ???, battery voltage %d mV, charging = %s, charge status %d = %s, charge level %s, charge type %s',
voltage, status, (flags & 0x03), charge_sts, charge_lvl, charge_type)
return charge_lvl, status, voltage, charge_sts, charge_type
@ -635,10 +634,9 @@ def get_hi_res_scrolling_info(device):
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
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

4
lib/logitech_receiver/i18n.py
View File

@ -24,10 +24,10 @@ from __future__ import absolute_import, division, print_function, unicode_litera
import gettext as _gettext
try:
unicode
unicode # noqa: F821
_ = lambda x: _gettext.gettext(x).decode('UTF-8')
ngettext = lambda *x: _gettext.ngettext(*x).decode('UTF-8')
except:
except Exception:
_ = _gettext.gettext
ngettext = _gettext.ngettext

38
lib/logitech_receiver/listener.py
View File

@ -46,8 +46,8 @@ del getLogger
class _ThreadedHandle(object):
"""A thread-local wrapper with different open handles for each thread.
Closing a ThreadedHandle will close all handles.
"""
Closing a ThreadedHandle will close all handles.
"""
__slots__ = ('path', '_local', '_handles', '_listener')
@ -70,7 +70,7 @@ class _ThreadedHandle(object):
_log.error('%r failed to open new handle', self)
else:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("%r opened new handle %d", self, handle)
# _log.debug("%r opened new handle %d", self, handle)
self._local.handle = handle
self._handles.append(handle)
return handle
@ -99,7 +99,7 @@ class _ThreadedHandle(object):
if self._local:
try:
return self._local.handle
except:
except Exception:
return self._open()
__int__ = __index__
@ -139,8 +139,8 @@ _EVENT_READ_TIMEOUT = 0.4 # in seconds
class EventsListener(_threading.Thread):
"""Listener thread for notifications from the Unifying Receiver.
Incoming packets will be passed to the callback function in sequence.
"""
Incoming packets will be passed to the callback function in sequence.
"""
def __init__(self, receiver, notifications_callback):
super(EventsListener, self).__init__(name=self.__class__.__name__ +
':' + receiver.path.split('/')[2])
@ -190,20 +190,20 @@ class EventsListener(_threading.Thread):
if n:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("%s: processing %s", self.receiver, n)
# _log.debug("%s: processing %s", self.receiver, n)
try:
self._notifications_callback(n)
except:
except Exception:
_log.exception('processing %s', n)
# elif self.tick_period:
# idle_reads -= 1
# if idle_reads <= 0:
# idle_reads = _IDLE_READS
# now = _timestamp()
# if now - last_tick >= self.tick_period:
# last_tick = now
# self.tick(now)
# idle_reads -= 1
# if idle_reads <= 0:
# idle_reads = _IDLE_READS
# now = _timestamp()
# if now - last_tick >= self.tick_period:
# last_tick = now
# self.tick(now)
del self._queued_notifications
self.has_stopped()
@ -214,7 +214,7 @@ class EventsListener(_threading.Thread):
def has_started(self):
"""Called right after the thread has started, and before it starts
reading notification packets."""
reading notification packets."""
pass
def has_stopped(self):
@ -222,8 +222,8 @@ class EventsListener(_threading.Thread):
pass
# def tick(self, timestamp):
# """Called about every tick_period seconds."""
# pass
# """Called about every tick_period seconds."""
# pass
def _notifications_hook(self, n):
# Only consider unhandled notifications that were sent from this thread,
@ -231,7 +231,7 @@ class EventsListener(_threading.Thread):
assert _threading.current_thread() == self
if self._active: # and _threading.current_thread() == self:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("queueing unhandled %s", n)
# _log.debug("queueing unhandled %s", n)
if not self._queued_notifications.full():
self._queued_notifications.put(n)

22
lib/logitech_receiver/receiver.py
View File

@ -81,7 +81,7 @@ class PairedDevice(object):
self._power_switch = None
# if _log.isEnabledFor(_DEBUG):
# _log.debug("new PairedDevice(%s, %s, %s)", receiver, number, link_notification)
# _log.debug("new PairedDevice(%s, %s, %s)", receiver, number, link_notification)
if link_notification is not None:
self.online = not bool(ord(link_notification.data[0:1]) & 0x40)
@ -156,7 +156,7 @@ class PairedDevice(object):
self.online = self._protocol is not None
# if _log.isEnabledFor(_DEBUG):
# _log.debug("device %d protocol %s", self.number, self._protocol)
# _log.debug("device %d protocol %s", self.number, self._protocol)
return self._protocol or 0
@property
@ -169,7 +169,7 @@ class PairedDevice(object):
codename = codename[2:2 + codename_length]
self._codename = codename.decode('ascii')
# if _log.isEnabledFor(_DEBUG):
# _log.debug("device %d codename %s", self.number, self._codename)
# _log.debug("device %d codename %s", self.number, self._codename)
else:
self._codename = '? (%s)' % self.wpid
return self._codename
@ -276,7 +276,7 @@ class PairedDevice(object):
def enable_notifications(self, enable=True):
"""Enable or disable device (dis)connection notifications on this
receiver."""
receiver."""
if not bool(self.receiver) or self.protocol >= 2.0:
return False
@ -350,8 +350,8 @@ class PairedDevice(object):
class Receiver(object):
"""A Unifying Receiver instance.
The paired devices are available through the sequence interface.
"""
The paired devices are available through the sequence interface.
"""
number = 0xFF
kind = None
@ -414,7 +414,7 @@ class Receiver(object):
def enable_notifications(self, enable=True):
"""Enable or disable device (dis)connection notifications on this
receiver."""
receiver."""
if not self.handle:
return False
@ -482,7 +482,7 @@ class Receiver(object):
return 0 if count is None else ord(count[1:2])
# def has_devices(self):
# return len(self) > 0 or self.count() > 0
# return len(self) > 0 or self.count() > 0
def request(self, request_id, *params):
if bool(self):
@ -579,8 +579,8 @@ class Receiver(object):
def open(self, device_info):
"""Opens a Logitech Receiver found attached to the machine, by Linux device path.
:returns: An open file handle for the found receiver, or ``None``.
"""
:returns: An open file handle for the found receiver, or ``None``.
"""
try:
handle = _base.open_path(device_info.path)
if handle:
@ -589,5 +589,5 @@ class Receiver(object):
_log.exception('open %s', device_info)
if e.errno == _errno.EACCES:
raise
except:
except Exception:
_log.exception('open %s', device_info)

25
lib/logitech_receiver/settings.py
View File

@ -23,7 +23,6 @@ import math
from copy import copy as _copy
from logging import DEBUG as _DEBUG
from logging import INFO as _INFO
from logging import getLogger
from .common import NamedInt as _NamedInt
@ -47,7 +46,7 @@ KIND = _NamedInts(toggle=0x01,
class Setting(object):
"""A setting descriptor.
Needs to be instantiated for each specific device."""
Needs to be instantiated for each specific device."""
__slots__ = ('name', 'label', 'description', 'kind', 'device_kind',
'feature', '_rw', '_validator', '_device', '_value')
@ -196,7 +195,7 @@ class Setting(object):
class Settings(Setting):
"""A setting descriptor for multiple choices, being a map from keys to values.
Needs to be instantiated for each specific device."""
Needs to be instantiated for each specific device."""
def read(self, cached=True):
assert hasattr(self, '_value')
assert hasattr(self, '_device')
@ -220,7 +219,7 @@ class Settings(Setting):
if self._device.online:
reply_map = {}
for key, value in self._validator.choices.items():
for key in self._validator.choices:
reply = self._rw.read(self._device, key)
if reply:
# keys are ints, because that is what the device uses,
@ -327,7 +326,7 @@ class Settings(Setting):
class BitFieldSetting(Setting):
"""A setting descriptor for a set of choices represented by one bit each, being a map from options to booleans.
Needs to be instantiated for each specific device."""
Needs to be instantiated for each specific device."""
def read(self, cached=True):
assert hasattr(self, '_value')
assert hasattr(self, '_device')
@ -666,7 +665,7 @@ class BitFieldValidator(object):
r = _bytes2int(reply_bytes[:self.byte_count])
value = {str(int(k)): False for k in self.options}
m = 1
for i in range(8 * self.byte_count):
for _ in range(8 * self.byte_count):
if m in self.options:
value[str(int(m))] = bool(r & m)
m <<= 1
@ -686,9 +685,9 @@ class ChoicesValidator(object):
kind = KIND.choice
"""Translates between NamedInts and a byte sequence.
:param choices: a list of NamedInts
:param bytes_count: the size of the derived byte sequence. If None, it
will be calculated from the choices."""
:param choices: a list of NamedInts
:param bytes_count: the size of the derived byte sequence. If None, it
will be calculated from the choices."""
def __init__(self, choices, bytes_count=None):
assert choices is not None
assert isinstance(choices, _NamedInts)
@ -790,10 +789,10 @@ class RangeValidator(object):
kind = KIND.range
"""Translates between integers and a byte sequence.
:param min_value: minimum accepted value (inclusive)
:param max_value: maximum accepted value (inclusive)
:param bytes_count: the size of the derived byte sequence. If None, it
will be calculated from the range."""
:param min_value: minimum accepted value (inclusive)
:param max_value: maximum accepted value (inclusive)
:param bytes_count: the size of the derived byte sequence. If None, it
will be calculated from the range."""
def __init__(self, min_value, max_value, bytes_count=None):
assert max_value > min_value
self.min_value = min_value

16
lib/logitech_receiver/settings_templates.py
View File

@ -28,7 +28,6 @@ from . import hidpp20 as _hidpp20
from . import special_keys as _special_keys
from .common import NamedInt as _NamedInt
from .common import NamedInts as _NamedInts
from .common import ReprogrammableKeyInfoV4 as _ReprogrammableKeyInfoV4
from .common import bytes2int as _bytes2int
from .common import int2bytes as _int2bytes
from .common import unpack as _unpack
@ -449,7 +448,8 @@ def _feature_k375s_fn_swap():
device_kind=(_DK.keyboard, ))
# FIXME: This will enable all supported backlight settings, we should allow the users to select which settings they want to enable.
# FIXME: This will enable all supported backlight settings,
# we should allow the users to select which settings they want to enable.
def _feature_backlight2():
return feature_toggle(_BACKLIGHT[0],
_F.BACKLIGHT2,
@ -729,11 +729,11 @@ def check_feature_settings(device, already_known):
def check_feature(name, featureId, featureFn):
"""
:param name: name for the setting
:param featureId: the numeric Feature ID for this setting implementation
:param featureFn: the function for this setting implementation
"""
if not featureId in device.features:
:param name: name for the setting
:param featureId: the numeric Feature ID for this setting implementation
:param featureFn: the function for this setting implementation
"""
if featureId not in device.features:
return
if any(s.name == name for s in already_known):
return
@ -749,7 +749,7 @@ def check_feature_settings(device, already_known):
_log.error('check_feature[%s] inconsistent feature %s', featureId,
reason)
for name, featureId, featureFn, _, _ in _SETTINGS_TABLE:
for name, featureId, featureFn, __, __ in _SETTINGS_TABLE:
if featureId and featureFn:
check_feature(name, featureId, featureFn)
return True

6
lib/logitech_receiver/special_keys.py
View File

@ -194,7 +194,8 @@ CONTROL = _NamedInts(
Fn_Left_Click=0x00B7, # from K400 Plus
# https://docs.google.com/document/u/0/d/1YvXICgSe8BcBAuMr4Xu_TutvAxaa-RnGfyPFWBWzhkc/export?format=docx
# Extract to csv. Eliminate extra linefeeds and spaces.
# awk -F, '/0x/{gsub(" \\+ ","_",$2); gsub("/","__",$2); gsub(" -","_Down",$2); gsub(" \\+","_Up",$2); gsub("[()\"-]","",$2); gsub(" ","_",$2); printf("\t%s=0x%04X,\n", $2, $1)}' < controls.cvs
# awk -F, '/0x/{gsub(" \\+ ","_",$2); gsub("/","__",$2); gsub(" -","_Down",$2);
# gsub(" \\+","_Up",$2); gsub("[()\"-]","",$2); gsub(" ","_",$2); printf("\t%s=0x%04X,\n", $2, $1)}' < controls.cvs
Second_Left_Click=0x00B8, # Second_LClick / on K400 Plus
Fn_Second_Left_Click=0x00B9, # Fn_Second_LClick
MultiPlatform_App_Switch=0x00BA,
@ -415,7 +416,8 @@ TASK = _NamedInts(
ShowUI=0x0092,
# https://docs.google.com/document/d/1Dpx_nWRQAZox_zpZ8SNc9nOkSDE9svjkghOCbzopabc/edit
# Extract to csv. Eliminate extra linefeeds and spaces. Turn / into __ and space into _
# awk -F, '/0x/{gsub(" \\+ ","_",$2); gsub("_-","_Down",$2); gsub("_\\+","_Up",$2); gsub("[()\"-]","",$2); gsub(" ","_",$2); printf("\t%s=0x%04X,\n", $2, $1)}' < tasks.csv > tasks.py
# awk -F, '/0x/{gsub(" \\+ ","_",$2); gsub("_-","_Down",$2); gsub("_\\+","_Up",$2);
# gsub("[()\"-]","",$2); gsub(" ","_",$2); printf("\t%s=0x%04X,\n", $2, $1)}' < tasks.csv > tasks.py
Switch_Presentation__Switch_Screen=0x0093, # on K400 Plus
Minimize_Window=0x0094,
Maximize_Window=0x0095, # on K400 Plus

87
lib/logitech_receiver/status.py
View File

@ -88,8 +88,8 @@ def attach_to(device, changed_callback):
class ReceiverStatus(dict):
"""The 'runtime' status of a receiver, mostly about the pairing process --
is the pairing lock open or closed, any pairing errors, etc.
"""
is the pairing lock open or closed, any pairing errors, etc.
"""
def __init__(self, receiver, changed_callback):
assert receiver
self._receiver = receiver
@ -118,17 +118,17 @@ class ReceiverStatus(dict):
self._changed_callback(self._receiver, alert=alert, reason=reason)
# def poll(self, timestamp):
# r = self._receiver
# assert r
# r = self._receiver
# assert r
#
# if _log.isEnabledFor(_DEBUG):
# _log.debug("polling status of %s", r)
# if _log.isEnabledFor(_DEBUG):
# _log.debug("polling status of %s", r)
#
# # make sure to read some stuff that may be read later by the UI
# r.serial, r.firmware, None
# # make sure to read some stuff that may be read later by the UI
# r.serial, r.firmware, None
#
# # get an update of the notification flags
# # self[KEYS.NOTIFICATION_FLAGS] = _hidpp10.get_notification_flags(r)
# # get an update of the notification flags
# # self[KEYS.NOTIFICATION_FLAGS] = _hidpp10.get_notification_flags(r)
#
@ -138,9 +138,9 @@ class ReceiverStatus(dict):
class DeviceStatus(dict):
"""Holds the 'runtime' status of a peripheral -- things like
active/inactive, battery charge, lux, etc. It updates them mostly by
processing incoming notification events from the device itself.
"""
active/inactive, battery charge, lux, etc. It updates them mostly by
processing incoming notification events from the device itself.
"""
def __init__(self, device, changed_callback):
assert device
self._device = device
@ -177,7 +177,8 @@ class DeviceStatus(dict):
light_level = self.get(KEYS.LIGHT_LEVEL)
if light_level is not None:
if comma: yield ', '
if comma:
yield ', '
yield _('Lighting: %(level)s lux') % {'level': light_level}
return ''.join(i for i in _items())
@ -352,7 +353,7 @@ class DeviceStatus(dict):
if battery is not None:
self[KEYS.BATTERY_LEVEL] = battery
if self.updated == 0 and active == True:
if self.updated == 0 and active is True:
# if the device is active on the very first status notification,
# (meaning just when the program started or a new receiver was just
# detected), pop-up a notification about it
@ -360,40 +361,40 @@ class DeviceStatus(dict):
self.updated = timestamp
# if _log.isEnabledFor(_DEBUG):
# _log.debug("device %d changed: active=%s %s", d.number, self._active, dict(self))
# _log.debug("device %d changed: active=%s %s", d.number, self._active, dict(self))
self._changed_callback(d, alert, reason)
# def poll(self, timestamp):
# d = self._device
# if not d:
# _log.error("polling status of invalid device")
# return
# d = self._device
# if not d:
# _log.error("polling status of invalid device")
# return
#
# if self._active:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("polling status of %s", d)
# if self._active:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("polling status of %s", d)
#
# # read these from the device, the UI may need them later
# d.protocol, d.serial, d.firmware, d.kind, d.name, d.settings, None
# # read these from the device, the UI may need them later
# d.protocol, d.serial, d.firmware, d.kind, d.name, d.settings, None
#
# # make sure we know all the features of the device
# # if d.features:
# # d.features[:]
# # make sure we know all the features of the device
# # if d.features:
# # d.features[:]
#
# # devices may go out-of-range while still active, or the computer
# # may go to sleep and wake up without the devices available
# if timestamp - self.updated > _STATUS_TIMEOUT:
# if d.ping():
# timestamp = self.updated = _timestamp()
# else:
# self.changed(active=False, reason='out of range')
# # devices may go out-of-range while still active, or the computer
# # may go to sleep and wake up without the devices available
# if timestamp - self.updated > _STATUS_TIMEOUT:
# if d.ping():
# timestamp = self.updated = _timestamp()
# else:
# self.changed(active=False, reason='out of range')
#
# # if still active, make sure we know the battery level
# if KEYS.BATTERY_LEVEL not in self:
# self.read_battery(timestamp)
# # if still active, make sure we know the battery level
# if KEYS.BATTERY_LEVEL not in self:
# self.read_battery(timestamp)
#
# elif timestamp - self.updated > _STATUS_TIMEOUT:
# if d.ping():
# self.changed(active=True)
# else:
# self.updated = _timestamp()
# elif timestamp - self.updated > _STATUS_TIMEOUT:
# if d.ping():
# self.changed(active=True)
# else:
# self.updated = _timestamp()

14
lib/solaar/cli/__init__.py
View File

@ -139,11 +139,12 @@ def _find_device(receivers, name):
if len(name) == 1:
try:
number = int(name)
except:
except Exception:
pass
else:
assert not (number < 0)
if number > 6: number = None
if number > 6:
number = None
for r in receivers:
if number and number <= r.max_devices:
@ -168,7 +169,7 @@ def run(cli_args=None, hidraw_path=None):
args = _cli_parser.parse_args()
# Python 3 has an undocumented 'feature' that breaks parsing empty args
# http://bugs.python.org/issue16308
if not 'cmd' in args:
if 'cmd' not in args:
_cli_parser.print_usage(_sys.stderr)
_sys.stderr.write('%s: error: too few arguments\n' % NAME.lower())
_sys.exit(2)
@ -183,11 +184,10 @@ def run(cli_args=None, hidraw_path=None):
from importlib import import_module
m = import_module('.' + action, package=__name__)
m.run(c, args, _find_receiver, _find_device)
except AssertionError as e:
except AssertionError:
from traceback import extract_tb
tb_last = extract_tb(_sys.exc_info()[2])[-1]
_sys.exit('%s: assertion failed: %s line %d' %
(NAME.lower(), tb_last[0], tb_last[1]))
except Exception as e:
_sys.exit('%s: assertion failed: %s line %d' % (NAME.lower(), tb_last[0], tb_last[1]))
except Exception:
from traceback import format_exc
_sys.exit('%s: error: %s' % (NAME.lower(), format_exc()))

2
lib/solaar/cli/config.py
View File

@ -84,7 +84,7 @@ def run(receivers, args, find_receiver, find_device):
value = args.value
try:
value = bool(int(value))
except:
except Exception:
if value.lower() in ('true', 'yes', 'on', 't', 'y'):
value = True
elif value.lower() in ('false', 'no', 'off', 'f', 'n'):

5
lib/solaar/cli/probe.py
View File

@ -19,12 +19,7 @@
from __future__ import absolute_import, division, print_function, unicode_literals
from time import time as _timestamp
from logitech_receiver import base as _base
from logitech_receiver import hidpp10 as _hidpp10
from logitech_receiver import notifications as _notifications
from logitech_receiver import status as _status
from logitech_receiver.common import strhex as _strhex
from solaar.cli.show import _print_receiver

6
lib/solaar/configuration.py
View File

@ -46,7 +46,7 @@ def _load():
try:
with open(_file_path, 'r') as config_file:
loaded_configuration = _json_load(config_file)
except:
except Exception:
_log.error('failed to load from %s', _file_path)
# loaded_configuration.update(_configuration)
@ -70,7 +70,7 @@ def save():
if not _path.isdir(dirname):
try:
_os.makedirs(dirname)
except:
except Exception:
_log.error('failed to create %s', dirname)
return False
@ -87,7 +87,7 @@ def save():
if _log.isEnabledFor(_INFO):
_log.info('saved %s to %s', _configuration, _file_path)
return True
except:
except Exception:
_log.error('failed to save to %s', _file_path)

8
lib/solaar/gtk.py
View File

@ -129,7 +129,8 @@ def main():
signal.signal(signal.SIGINT, signal.SIG_DFL)
args = _parse_arguments()
if not args: return
if not args:
return
if args.action:
# if any argument, run comandline and exit
return _cli.run(args.action, args.hidraw_path)
@ -149,9 +150,8 @@ def main():
_upower.watch(lambda: listener.ping_all(True))
# main UI event loop
ui.run_loop(listener.start_all, listener.stop_all,
args.window != 'only', args.window != 'hide')
except Exception as e:
ui.run_loop(listener.start_all, listener.stop_all, args.window != 'only', args.window != 'hide')
except Exception:
import sys
from traceback import format_exc
sys.exit('%s: error: %s' % (NAME.lower(), format_exc()))

4
lib/solaar/i18n.py
View File

@ -63,9 +63,9 @@ _gettext.textdomain(_LOCALE_DOMAIN)
_gettext.install(_LOCALE_DOMAIN)
try:
unicode
unicode # noqa: F821
_ = lambda x: _gettext.gettext(x).decode('UTF-8')
ngettext = lambda *x: _gettext.ngettext(*x).decode('UTF-8')
except:
except Exception:
_ = _gettext.gettext
ngettext = _gettext.ngettext

75
lib/solaar/listener.py
View File

@ -70,7 +70,7 @@ def _ghost(device):
class ReceiverListener(_listener.EventsListener):
"""Keeps the status of a Receiver.
"""
"""
def __init__(self, receiver, status_changed_callback):
super(ReceiverListener, self).__init__(receiver,
self._notifications_handler)
@ -90,7 +90,7 @@ class ReceiverListener(_listener.EventsListener):
self.receiver.status[
_status.KEYS.NOTIFICATION_FLAGS] = notification_flags
self.receiver.notify_devices()
self._status_changed(self.receiver) #, _status.ALERT.NOTIFICATION)
self._status_changed(self.receiver) # , _status.ALERT.NOTIFICATION)
def has_stopped(self):
r, self.receiver = self.receiver, None
@ -106,46 +106,46 @@ class ReceiverListener(_listener.EventsListener):
if r:
try:
r.close()
except:
except Exception:
_log.exception('closing receiver %s' % r.path)
self.status_changed_callback(r) #, _status.ALERT.NOTIFICATION)
self.status_changed_callback(r) # , _status.ALERT.NOTIFICATION)
# def tick(self, timestamp):
# if not self.tick_period:
# raise Exception("tick() should not be called without a tick_period: %s", self)
# if not self.tick_period:
# raise Exception("tick() should not be called without a tick_period: %s", self)
#
# # not necessary anymore, we're now using udev monitor to watch for receiver status
# # if self._last_tick > 0 and timestamp - self._last_tick > _POLL_TICK * 2:
# # # if we missed a couple of polls, most likely the computer went into
# # # sleep, and we have to reinitialize the receiver again
# # _log.warn("%s: possible sleep detected, closing this listener", self.receiver)
# # self.stop()
# # return
# # not necessary anymore, we're now using udev monitor to watch for receiver status
# # if self._last_tick > 0 and timestamp - self._last_tick > _POLL_TICK * 2:
# # # if we missed a couple of polls, most likely the computer went into
# # # sleep, and we have to reinitialize the receiver again
# # _log.warn("%s: possible sleep detected, closing this listener", self.receiver)
# # self.stop()
# # return
#
# self._last_tick = timestamp
# self._last_tick = timestamp
#
# try:
# # read these in case they haven't been read already
# # self.receiver.serial, self.receiver.firmware
# if self.receiver.status.lock_open:
# # don't mess with stuff while pairing
# return
# try:
# # read these in case they haven't been read already
# # self.receiver.serial, self.receiver.firmware
# if self.receiver.status.lock_open:
# # don't mess with stuff while pairing
# return
#
# self.receiver.status.poll(timestamp)
# self.receiver.status.poll(timestamp)
#
# # Iterating directly through the reciver would unnecessarily probe
# # all possible devices, even unpaired ones.
# # Checking for each device number in turn makes sure only already
# # known devices are polled.
# # This is okay because we should have already known about them all
# # long before the first poll() happents, through notifications.
# for number in range(1, 6):
# if number in self.receiver:
# dev = self.receiver[number]
# if dev and dev.status is not None:
# dev.status.poll(timestamp)
# except Exception as e:
# _log.exception("polling", e)
# # Iterating directly through the reciver would unnecessarily probe
# # all possible devices, even unpaired ones.
# # Checking for each device number in turn makes sure only already
# # known devices are polled.
# # This is okay because we should have already known about them all
# # long before the first poll() happents, through notifications.
# for number in range(1, 6):
# if number in self.receiver:
# dev = self.receiver[number]
# if dev and dev.status is not None:
# dev.status.poll(timestamp)
# except Exception as e:
# _log.exception("polling", e)
def _status_changed(self, device, alert=_status.ALERT.NONE, reason=None):
assert device is not None
@ -184,7 +184,7 @@ class ReceiverListener(_listener.EventsListener):
def _notifications_handler(self, n):
assert self.receiver
# if _log.isEnabledFor(_DEBUG):
# _log.debug("%s: handling %s", self.receiver, n)
# _log.debug("%s: handling %s", self.receiver, n)
if n.devnumber == 0xFF:
# a receiver notification
_notifications.process(self.receiver, n)
@ -370,10 +370,11 @@ def _process_receiver_event(action, device_info):
# (It would be easier to use pylibacl but adding the pylibacl dependencies
# for this special case is not good.)
try:
import subprocess, re
import subprocess
import re
output = subprocess.check_output(
['/usr/bin/getfacl', '-p', device_info.path])
if not re.search(b'user:.+:', output):
_error_callback('permissions', device_info.path)
except:
except Exception:
_error_callback('permissions', device_info.path)

2
lib/solaar/tasks.py
View File

@ -65,7 +65,7 @@ class TaskRunner(_Thread):
assert function
try:
function(*args, **kwargs)
except:
except Exception:
_log.exception('calling %s', function)
if _log.isEnabledFor(_DEBUG):

10
lib/solaar/ui/__init__.py
View File

@ -47,14 +47,12 @@ def _error_dialog(reason, object):
if reason == 'permissions':
title = _('Permissions error')
text = _('Found a Logitech Receiver (%s), but did not have permission to open it.') % object + \
'\n\n' + \
_("If you've just installed Solaar, try removing the receiver and plugging it back in.")
text = (_('Found a Logitech Receiver (%s), but did not have permission to open it.') % object + '\n\n' +
_("If you've just installed Solaar, try removing the receiver and plugging it back in."))
elif reason == 'unpair':
title = _('Unpairing failed')
text = _('Failed to unpair %{device} from %{receiver}.').format(device=object.name, receiver=object.receiver.name) + \
'\n\n' + \
_('The receiver returned an error, with no further details.')
text = (_('Failed to unpair %{device} from %{receiver}.').format(device=object.name, receiver=object.receiver.name) +
'\n\n' + _('The receiver returned an error, with no further details.'))
else:
raise Exception("ui.error_dialog: don't know how to handle (%s, %s)",
reason, object)

2
lib/solaar/ui/about.py
View File

@ -61,7 +61,7 @@ def _create():
# gtk3 < ~3.6.4 has incorrect gi bindings
import logging
logging.exception('failed to fully create the about dialog')
except:
except Exception:
# the Gtk3 version may be too old, and the function does not exist
import logging
logging.exception('failed to fully create the about dialog')

12
lib/solaar/ui/action.py
View File

@ -60,11 +60,11 @@ def make_toggle(name, label, function, stock_id=None, *args):
#
# def _toggle_notifications(action):
# if action.get_active():
# notify.init('Solaar')
# else:
# notify.uninit()
# action.set_sensitive(notify.available)
# if action.get_active():
# notify.init('Solaar')
# else:
# notify.uninit()
# action.set_sensitive(notify.available)
# toggle_notifications = make_toggle('notifications', 'Notifications', _toggle_notifications)
about = make('help-about',
@ -109,6 +109,6 @@ def unpair(window, device):
try:
del receiver[device_number]
except:
except Exception:
# _log.exception("unpairing %s", device)
error_dialog('unpair', device)

6
lib/solaar/ui/config_panel.py
View File

@ -75,7 +75,7 @@ def _write_async_key_value(setting, key, value, sbox):
def _create_toggle_control(setting):
def _switch_notify(switch, _ignore, s):
if switch.get_sensitive():
_write_async(s, switch.get_active() == True, switch.get_parent())
_write_async(s, switch.get_active() is True, switch.get_parent())
c = Gtk.Switch()
c.connect('notify::active', _switch_notify, setting)
@ -312,8 +312,8 @@ def update(device, is_online=None):
def clean(device):
"""Remove the controls for a given device serial.
Needed after the device has been unpaired.
"""
Needed after the device has been unpaired.
"""
assert _box is not None
device_id = (device.receiver.path, device.number)
for k in list(_items.keys()):

2
lib/solaar/ui/icons.py
View File

@ -219,7 +219,7 @@ def icon_file(name, size=_LARGE_SIZE):
if theme_icon:
file_name = theme_icon.get_filename()
# if _log.isEnabledFor(_DEBUG):
# _log.debug("icon %s(%d) => %s", name, size, file_name)
# _log.debug("icon %s(%d) => %s", name, size, file_name)
return file_name
_log.warn('icon %s(%d) not found in current theme', name, size)

24
lib/solaar/ui/notify.py
View File

@ -60,7 +60,7 @@ if available:
_log.info('starting desktop notifications')
try:
return Notify.init(NAME)
except:
except Exception:
_log.exception('initializing desktop notifications')
available = False
return available and Notify.is_initted()
@ -73,12 +73,12 @@ if available:
Notify.uninit()
# def toggle(action):
# if action.get_active():
# init()
# else:
# uninit()
# action.set_sensitive(available)
# return action.get_active()
# if action.get_active():
# init()
# else:
# uninit()
# action.set_sensitive(available)
# return action.get_active()
def alert(reason, icon=None):
assert reason
@ -90,8 +90,7 @@ if available:
# we need to use the filename here because the notifications daemon
# is an external application that does not know about our icon sets
icon_file = _icons.icon_file(NAME.lower()) if icon is None \
else _icons.icon_file(icon)
icon_file = _icons.icon_file(NAME.lower()) if icon is None else _icons.icon_file(icon)
n.update(NAME, reason, icon_file)
n.set_urgency(Notify.Urgency.NORMAL)
@ -99,7 +98,7 @@ if available:
try:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("showing %s", n)
# _log.debug("showing %s", n)
n.show()
except Exception:
_log.exception('showing %s', n)
@ -125,8 +124,7 @@ if available:
# we need to use the filename here because the notifications daemon
# is an external application that does not know about our icon sets
icon_file = _icons.device_icon_file(dev.name, dev.kind) if icon is None \
else _icons.icon_file(icon)
icon_file = _icons.device_icon_file(dev.name, dev.kind) if icon is None else _icons.icon_file(icon)
n.update(summary, message, icon_file)
urgency = Notify.Urgency.LOW if dev.status else Notify.Urgency.NORMAL
@ -135,7 +133,7 @@ if available:
try:
# if _log.isEnabledFor(_DEBUG):
# _log.debug("showing %s", n)
# _log.debug("showing %s", n)
n.show()
except Exception:
_log.exception('showing %s', n)

11
lib/solaar/ui/pair_window.py
View File

@ -184,14 +184,9 @@ def _pairing_succeeded(assistant, receiver, device):
def _check_encrypted(dev):
if assistant.is_drawable():
if device.status.get(_K.LINK_ENCRYPTED) == False:
hbox.pack_start(
Gtk.Image.new_from_icon_name('security-low',
Gtk.IconSize.MENU), False,
False, 0)
hbox.pack_start(
Gtk.Label(_('The wireless link is not encrypted') + '!'),
False, False, 0)
if device.status.get(_K.LINK_ENCRYPTED) is False:
hbox.pack_start(Gtk.Image.new_from_icon_name('security-low', Gtk.IconSize.MENU), False, False, 0)
hbox.pack_start(Gtk.Label(_('The wireless link is not encrypted') + '!'), False, False, 0)
hbox.show_all()
else:
return True

4
lib/solaar/ui/tray.py
View File

@ -42,7 +42,7 @@ del getLogger
# constants
#
_TRAY_ICON_SIZE = 32 # pixels
_TRAY_ICON_SIZE = 32 # pixels
_MENU_ICON_SIZE = Gtk.IconSize.LARGE_TOOLBAR
_RECEIVER_SEPARATOR = ('~', None, None, None)
@ -103,7 +103,7 @@ def _scroll(tray_icon, event, direction=None):
_last_scroll = now
# if _log.isEnabledFor(_DEBUG):
# _log.debug("scroll direction %s", direction)
# _log.debug("scroll direction %s", direction)
global _picked_device
candidate = None

8
lib/solaar/ui/window.py
View File

@ -415,7 +415,7 @@ def _device_selected(selection):
model, item = selection.get_selected()
device = model.get_value(item, _COLUMN.DEVICE) if item else None
# if _log.isEnabledFor(_DEBUG):
# _log.debug("window tree selected device %s", device)
# _log.debug("window tree selected device %s", device)
if device:
_update_info_panel(device, full=True)
else:
@ -666,8 +666,8 @@ def _update_receiver_panel(receiver, panel, buttons, full=False):
# b._insecure.set_visible(False)
buttons._unpair.set_visible(False)
if ( receiver.may_unpair or receiver.re_pairs ) and not is_pairing and \
( receiver.remaining_pairings() is None or receiver.remaining_pairings() != 0 ):
if (receiver.may_unpair or receiver.re_pairs) and not is_pairing and \
(receiver.remaining_pairings() is None or receiver.remaining_pairings() != 0):
if not receiver.re_pairs and devices_count >= receiver.max_devices:
paired_devices = tuple(n
for n in range(1, receiver.max_devices + 1)
@ -727,7 +727,7 @@ def _update_device_panel(device, panel, buttons, full=False):
panel._battery._text.set_markup(text)
if is_online:
not_secure = device.status.get(_K.LINK_ENCRYPTED) == False
not_secure = device.status.get(_K.LINK_ENCRYPTED) is False
if not_secure:
panel._secure._text.set_text(_('not encrypted'))
panel._secure._icon.set_from_icon_name('security-low',

4
lib/solaar/upower.py
View File

@ -55,7 +55,7 @@ def _suspend_or_resume(suspend):
def watch(on_resume_callback=None, on_suspend_callback=None):
"""Register callback for suspend/resume events.
They are called only if the system DBus is running, and the UPower daemon is available."""
They are called only if the system DBus is running, and the UPower daemon is available."""
global _resume_callback, _suspend_callback
_suspend_callback = on_suspend_callback
_resume_callback = on_resume_callback
@ -95,7 +95,7 @@ try:
_log.info(
'connected to system dbus, watching for suspend/resume events')
except:
except Exception:
# Either:
# - the dbus library is not available
# - the system dbus is not running

2
setup.py
View File

@ -9,7 +9,7 @@ except ImportError:
autostart_path = '/etc/xdg/autostart'
#from solaar import NAME, __version__
# from solaar import NAME, __version__
__version__ = '1.0.2'
NAME = 'Solaar'