device: streamline status code

This commit is contained in:
Peter F. Patel-Schneider 2024-02-17 12:44:26 -05:00
parent 50ddb54466
commit fa9494435b
2 changed files with 18 additions and 117 deletions

View File

@ -18,8 +18,6 @@
import logging import logging
from time import time as _timestamp
from . import hidpp10 as _hidpp10 from . import hidpp10 as _hidpp10
from . import hidpp20 as _hidpp20 from . import hidpp20 as _hidpp20
from . import settings as _settings from . import settings as _settings
@ -54,15 +52,6 @@ KEYS = _NamedInts(
# (blink systray icon/notification/whatever). # (blink systray icon/notification/whatever).
_BATTERY_ATTENTION_LEVEL = 5 _BATTERY_ATTENTION_LEVEL = 5
# If no updates have been receiver from the device for a while, ping the device
# and update it status accordingly.
# _STATUS_TIMEOUT = 5 * 60 # seconds
_LONG_SLEEP = 15 * 60 # seconds
#
#
#
def attach_to(device, changed_callback): def attach_to(device, changed_callback):
assert device assert device
@ -92,8 +81,6 @@ class ReceiverStatus(dict):
assert changed_callback assert changed_callback
self._changed_callback = changed_callback self._changed_callback = changed_callback
# self.updated = 0
self.lock_open = False self.lock_open = False
self.discovering = False self.discovering = False
self.counter = None self.counter = None
@ -116,22 +103,8 @@ class ReceiverStatus(dict):
) )
def changed(self, alert=ALERT.NOTIFICATION, reason=None): def changed(self, alert=ALERT.NOTIFICATION, reason=None):
# self.updated = _timestamp()
self._changed_callback(self._receiver, alert=alert, reason=reason) self._changed_callback(self._receiver, alert=alert, reason=reason)
# def poll(self, timestamp):
# r = self._receiver
# assert r
#
# if logger.isEnabledFor(logging.DEBUG):
# logger.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
#
# # get an update of the notification flags
# # self[KEYS.NOTIFICATION_FLAGS] = _hidpp10.get_notification_flags(r)
# #
# #
@ -147,41 +120,23 @@ class DeviceStatus(dict):
def __init__(self, device, changed_callback): def __init__(self, device, changed_callback):
assert device assert device
self._device = device self._device = device
assert changed_callback assert changed_callback
self._changed_callback = changed_callback self._changed_callback = changed_callback
self._active = None # is the device active?
# is the device active?
self._active = None
# timestamp of when this status object was last updated
self.updated = 0
def to_string(self): def to_string(self):
def _items(): status = ''
comma = False battery_level = self.get(KEYS.BATTERY_LEVEL)
if battery_level is not None:
battery_level = self.get(KEYS.BATTERY_LEVEL) if isinstance(battery_level, _NamedInt):
if battery_level is not None: status = _('Battery: %(level)s') % {'level': _(str(battery_level))}
if isinstance(battery_level, _NamedInt): else:
yield _('Battery: %(level)s') % {'level': _(str(battery_level))} status = _('Battery: %(percent)d%%') % {'percent': battery_level}
else: battery_status = self.get(KEYS.BATTERY_STATUS)
yield _('Battery: %(percent)d%%') % {'percent': battery_level} if battery_status is not None:
status += ' (%s)' % _(str(battery_status))
battery_status = self.get(KEYS.BATTERY_STATUS) return status
if battery_status is not None:
yield ' (%s)' % _(str(battery_status))
comma = True
light_level = self.get(KEYS.LIGHT_LEVEL)
if light_level is not None:
if comma:
yield ', '
yield _('Lighting: %(level)s lux') % {'level': light_level}
return ''.join(i for i in _items())
def __repr__(self): def __repr__(self):
return '{' + ', '.join('\'%s\': %r' % (k, v) for k, v in self.items()) + '}' return '{' + ', '.join('\'%s\': %r' % (k, v) for k, v in self.items()) + '}'
@ -191,7 +146,7 @@ class DeviceStatus(dict):
__nonzero__ = __bool__ __nonzero__ = __bool__
def set_battery_info(self, level, nextLevel, status, voltage, timestamp=None): def set_battery_info(self, level, nextLevel, status, voltage):
if logger.isEnabledFor(logging.DEBUG): if logger.isEnabledFor(logging.DEBUG):
logger.debug('%s: battery %s, %s', self._device, level, status) logger.debug('%s: battery %s, %s', self._device, level, status)
@ -241,10 +196,10 @@ class DeviceStatus(dict):
if changed or reason or not self._active: # a battery response means device is active if changed or reason or not self._active: # a battery response means device is active
# update the leds on the device, if any # update the leds on the device, if any
_hidpp10.set_3leds(self._device, level, charging=charging, warning=bool(alert)) _hidpp10.set_3leds(self._device, level, charging=charging, warning=bool(alert))
self.changed(active=True, alert=alert, reason=reason, timestamp=timestamp) self.changed(active=True, alert=alert, reason=reason)
# Retrieve and regularize battery status # Retrieve and regularize battery status
def read_battery(self, timestamp=None): def read_battery(self):
if self._active: if self._active:
assert self._device assert self._device
battery = self._device.battery() battery = self._device.battery()
@ -260,11 +215,10 @@ class DeviceStatus(dict):
self[KEYS.BATTERY_VOLTAGE] = None self[KEYS.BATTERY_VOLTAGE] = None
self.changed() self.changed()
def changed(self, active=None, alert=ALERT.NONE, reason=None, push=False, timestamp=None): def changed(self, active=None, alert=ALERT.NONE, reason=None, push=False):
assert self._changed_callback assert self._changed_callback
d = self._device d = self._device
# assert d # may be invalid when processing the 'unpaired' notification # assert d # may be invalid when processing the 'unpaired' notification
timestamp = timestamp or _timestamp()
if active is not None: if active is not None:
d.online = active d.online = active
@ -278,7 +232,7 @@ class DeviceStatus(dict):
self[KEYS.NOTIFICATION_FLAGS] = d.enable_connection_notifications() self[KEYS.NOTIFICATION_FLAGS] = d.enable_connection_notifications()
# battery information may have changed so try to read it now # battery information may have changed so try to read it now
self.read_battery(timestamp) self.read_battery()
# Push settings for new devices (was_active is None), # Push settings for new devices (was_active is None),
# when devices request software reconfiguration # when devices request software reconfiguration
@ -292,61 +246,8 @@ class DeviceStatus(dict):
else: else:
if was_active: # don't clear status when devices go inactive if was_active: # don't clear status when devices go inactive
## battery = self.get(KEYS.BATTERY_LEVEL)
## self.clear()
## # If we had a known battery level before, assume it's not going
## # to change much while the device is offline.
## if battery is not None:
## self[KEYS.BATTERY_LEVEL] = battery
pass pass
# A device that is not active on the first status notification
# but becomes active afterwards does not produce a pop-up notification
# so don't produce one here. This cuts off pop-ups when Solaar starts,
# which can be problematic if Solaar is autostarted.
## 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
## alert |= ALERT.NOTIFICATION
self.updated = timestamp
# if logger.isEnabledFor(logging.DEBUG): # if logger.isEnabledFor(logging.DEBUG):
# logger.debug("device %d changed: active=%s %s", d.number, self._active, dict(self)) # logger.debug("device %d changed: active=%s %s", d.number, self._active, dict(self))
self._changed_callback(d, alert, reason) self._changed_callback(d, alert, reason)
# def poll(self, timestamp):
# d = self._device
# if not d:
# logger.error("polling status of invalid device")
# return
#
# if self._active:
# if logger.isEnabledFor(logging.DEBUG):
# logger.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
#
# # 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')
#
# # 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()

View File

@ -670,7 +670,7 @@ def _update_device_panel(device, panel, buttons, full=False):
panel.set_sensitive(is_online) panel.set_sensitive(is_online)
if device.status.get(_K.BATTERY_LEVEL) is None: if device.status.get(_K.BATTERY_LEVEL) is None:
device.status.read_battery(device) device.status.read_battery()
battery_level = device.status.get(_K.BATTERY_LEVEL) battery_level = device.status.get(_K.BATTERY_LEVEL)
battery_voltage = device.status.get(_K.BATTERY_VOLTAGE) battery_voltage = device.status.get(_K.BATTERY_VOLTAGE)