device: use status attribute for error

This commit is contained in:
Peter F. Patel-Schneider 2024-03-06 17:51:04 -05:00
parent 8154cd759f
commit 135c8b8cb9
5 changed files with 21 additions and 29 deletions

View File

@ -33,7 +33,6 @@ from .common import Battery as _Battery
from .common import strhex as _strhex
from .i18n import _
from .status import ALERT as _ALERT
from .status import KEYS as _K
logger = logging.getLogger(__name__)
@ -69,12 +68,12 @@ def _process_receiver_notification(receiver, status, n):
reason = _("pairing lock is open") if status.lock_open else _("pairing lock is closed")
if logger.isEnabledFor(logging.INFO):
logger.info("%s: %s", receiver, reason)
status[_K.ERROR] = None
status.error = None
if status.lock_open:
status.new_device = None
pair_error = ord(n.data[:1])
if pair_error:
status[_K.ERROR] = error_string = _hidpp10_constants.PAIRING_ERRORS[pair_error]
status.error = error_string = _hidpp10_constants.PAIRING_ERRORS[pair_error]
status.new_device = None
logger.warning("pairing error %d: %s", pair_error, error_string)
status.changed(reason=reason)
@ -86,13 +85,13 @@ def _process_receiver_notification(receiver, status, n):
reason = _("discovery lock is open") if status.discovering else _("discovery lock is closed")
if logger.isEnabledFor(logging.INFO):
logger.info("%s: %s", receiver, reason)
status[_K.ERROR] = None
status.error = None
if status.discovering:
status.counter = status.device_address = status.device_authentication = status.device_name = None
status.device_passkey = None
discover_error = ord(n.data[:1])
if discover_error:
status[_K.ERROR] = discover_string = _hidpp10_constants.BOLT_PAIRING_ERRORS[discover_error]
status.error = discover_string = _hidpp10_constants.BOLT_PAIRING_ERRORS[discover_error]
logger.warning("bolt discovering error %d: %s", discover_error, discover_string)
status.changed(reason=reason)
return True
@ -120,7 +119,7 @@ def _process_receiver_notification(receiver, status, n):
reason = _("pairing lock is open") if status.lock_open else _("pairing lock is closed")
if logger.isEnabledFor(logging.INFO):
logger.info("%s: %s", receiver, reason)
status[_K.ERROR] = None
status.error = None
if not status.lock_open:
status.counter = status.device_address = status.device_authentication = status.device_name = None
pair_error = n.data[0]
@ -129,7 +128,7 @@ def _process_receiver_notification(receiver, status, n):
elif n.address == 0x02 and not pair_error:
status.new_device = receiver.register_new_device(n.data[7])
if pair_error:
status[_K.ERROR] = error_string = _hidpp10_constants.BOLT_PAIRING_ERRORS[pair_error]
status.error = error_string = _hidpp10_constants.BOLT_PAIRING_ERRORS[pair_error]
status.new_device = None
logger.warning("pairing error %d: %s", pair_error, error_string)
status.changed(reason=reason)
@ -230,7 +229,6 @@ def _process_hidpp10_notification(device, status, n):
if n.sub_id == 0x40: # device unpairing
if n.address == 0x02:
# device un-paired
status.clear()
device.wpid = None
device.status = None
if device.number in device.receiver:

View File

@ -32,8 +32,6 @@ _hidpp10 = hidpp10.Hidpp10()
ALERT = NamedInts(NONE=0x00, NOTIFICATION=0x01, SHOW_WINDOW=0x02, ATTENTION=0x04, ALL=0xFF)
KEYS = NamedInts(ERROR=7)
def attach_to(device, changed_callback):
assert device
@ -46,7 +44,7 @@ def attach_to(device, changed_callback):
device.status = DeviceStatus(device, changed_callback)
class ReceiverStatus(dict):
class ReceiverStatus:
"""The 'runtime' status of a receiver, mostly about the pairing process --
is the pairing lock open or closed, any pairing errors, etc.
"""
@ -57,6 +55,7 @@ class ReceiverStatus(dict):
assert changed_callback
self._changed_callback = changed_callback
self.notification_flags = None
self.error = None
self.lock_open = False
self.discovering = False
@ -68,8 +67,6 @@ class ReceiverStatus(dict):
self.device_passkey = None
self.new_device = None
self[KEYS.ERROR] = None
def to_string(self):
count = len(self._receiver)
return (
@ -85,9 +82,9 @@ class ReceiverStatus(dict):
self._changed_callback(self._receiver, alert=alert, reason=reason)
class DeviceStatus(dict):
class DeviceStatus:
"""Holds the 'runtime' status of a peripheral
Currently _active, battery, link_encrypted, notification_flags -- dict entries are being moved to attributs
Currently _active, battery, link_encrypted, notification_flags, error
Updates mostly come from incoming notification events from the device itself.
"""
@ -100,13 +97,11 @@ class DeviceStatus(dict):
self.battery = None
self.link_encrypted = None
self.notification_flags = None
self.error = None
def to_string(self):
return self.battery.to_str() if self.battery is not None else ""
def __repr__(self):
return "{" + ", ".join("'%s': %r" % (k, v) for k, v in self.items()) + "}"
def __bool__(self):
return bool(self._active)
@ -123,11 +118,11 @@ class DeviceStatus(dict):
alert, reason = ALERT.NONE, None
if info.ok():
self[KEYS.ERROR] = None
self.error = None
else:
logger.warning("%s: battery %d%%, ALERT %s", self._device, info.level, info.status)
if self.get(KEYS.ERROR) != info.status:
self[KEYS.ERROR] = info.status
if self.error != info.status:
self.error = info.status
alert = ALERT.NOTIFICATION | ALERT.ATTENTION
reason = info.to_str()

View File

@ -131,7 +131,7 @@ def run(receivers, args, find_receiver, _ignore):
dev = receiver.status.new_device
print("Paired device %d: %s (%s) [%s:%s]" % (dev.number, dev.name, dev.codename, dev.wpid, dev.serial))
else:
error = receiver.status.get(_status.KEYS.ERROR)
error = receiver.status.get(_status.error)
if error:
raise Exception("pairing failed: %s" % error)
else:

View File

@ -18,7 +18,6 @@ import logging
from gi.repository import GLib, Gtk
from logitech_receiver import hidpp10_constants as _hidpp10_constants
from logitech_receiver.status import KEYS as _K
from solaar.i18n import _, ngettext
@ -72,9 +71,10 @@ def _check_lock_state(assistant, receiver, count=2):
logger.debug("assistant %s destroyed, bailing out", assistant)
return False
if receiver.status.get(_K.ERROR):
if receiver.status.error:
# receiver.status.new_device = _fake_device(receiver)
_pairing_failed(assistant, receiver, receiver.status.pop(_K.ERROR))
_pairing_failed(assistant, receiver, receiver.status.error)
receiver.status.error = None
return False
if receiver.status.new_device:
@ -145,7 +145,7 @@ def _prepare(assistant, page, receiver):
if receiver.receiver_kind == "bolt":
if receiver.discover(timeout=_PAIRING_TIMEOUT):
assert receiver.status.new_device is None
assert receiver.status.get(_K.ERROR) is None
assert receiver.status.error is None
spinner = page.get_children()[-1]
spinner.start()
GLib.timeout_add(_STATUS_CHECK, _check_lock_state, assistant, receiver)
@ -154,7 +154,7 @@ def _prepare(assistant, page, receiver):
GLib.idle_add(_pairing_failed, assistant, receiver, "discovery did not start")
elif receiver.set_lock(False, timeout=_PAIRING_TIMEOUT):
assert receiver.status.new_device is None
assert receiver.status.get(_K.ERROR) is None
assert receiver.status.error is None
spinner = page.get_children()[-1]
spinner.start()
GLib.timeout_add(_STATUS_CHECK, _check_lock_state, assistant, receiver)
@ -178,7 +178,7 @@ def _finish(assistant, receiver):
if receiver.status.discovering:
receiver.discover(True)
if not receiver.status.lock_open and not receiver.status.discovering:
receiver.status[_K.ERROR] = None
receiver.status.error = None
def _pairing_failed(assistant, receiver, error):

View File

@ -190,7 +190,6 @@ def _create_buttons_box():
assert _find_selected_device_id() is not None
device = _find_selected_device()
assert device is not None
assert bool(device)
assert device.kind is not None
_action.unpair(_window, device)