added a devices status window
This commit is contained in:
Daniel Pavel
parent
4d63c960e5
commit
9ba6aa1c31
|
|
@ -0,0 +1,59 @@
|
|||
#
|
||||
#
|
||||
#
|
||||
|
||||
import threading
|
||||
from gi.repository import Gtk
|
||||
from gi.repository import GObject
|
||||
|
||||
from . import constants as C
|
||||
from .watcher import WatcherThread
|
||||
from . import ui
|
||||
|
||||
|
||||
def _status_updated(watcher, icon, window):
|
||||
while True:
|
||||
watcher.status_changed.wait()
|
||||
text = watcher.status_text
|
||||
watcher.status_changed.clear()
|
||||
|
||||
if icon:
|
||||
GObject.idle_add(icon.set_tooltip_text, text)
|
||||
|
||||
if window:
|
||||
ur_detected = watcher.has_receiver()
|
||||
devices = [ watcher.devices[k] for k in watcher.devices ] if ur_detected else []
|
||||
GObject.idle_add(ui.window.update, window, ur_detected, devices)
|
||||
|
||||
|
||||
# def _pair_new_device(trigger, watcher):
|
||||
# pass
|
||||
|
||||
|
||||
def run(images_path):
|
||||
GObject.threads_init()
|
||||
|
||||
ui.init(images_path)
|
||||
ui.notify.start(C.APP_TITLE, ui.image)
|
||||
|
||||
watcher = WatcherThread(ui.notify.show)
|
||||
watcher.start()
|
||||
|
||||
window = ui.window.create(C.APP_TITLE, ui.image)
|
||||
|
||||
menu_actions = [('Scan all devices', watcher.request_all_statuses),
|
||||
# ('Pair new device', _pair_new_device, watcher),
|
||||
None,
|
||||
('Quit', Gtk.main_quit)]
|
||||
|
||||
click_action = (ui.window.toggle, window) if window else None
|
||||
tray_icon = ui.icon.create(ui.image('icon'), C.APP_TITLE, menu_actions, click_action)
|
||||
|
||||
ui_update_thread = threading.Thread(target=_status_updated, name='ui_update', args=(watcher, tray_icon, window))
|
||||
ui_update_thread.daemon = True
|
||||
ui_update_thread.start()
|
||||
|
||||
Gtk.main()
|
||||
|
||||
watcher.stop()
|
||||
ui.notify.stop()
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
#
|
||||
# Commonly used strings
|
||||
#
|
||||
|
||||
APP_TITLE = 'Solaar'
|
||||
UNIFYING_RECEIVER = 'Unifying Receiver'
|
||||
NO_DEVICES = 'No devices attached.'
|
||||
SCANNING = 'Initializing...'
|
||||
NO_RECEIVER = 'Unifying Receiver not found.'
|
||||
FOUND_RECEIVER = 'Unifying Receiver found.'
|
||||
|
|
@ -0,0 +1,26 @@
|
|||
# pass
|
||||
|
||||
import os.path as _os_path
|
||||
from . import (icon, notify, window)
|
||||
|
||||
|
||||
_images_path = None
|
||||
_IMAGES = {}
|
||||
|
||||
|
||||
def init(images_path=None):
|
||||
global _images_path
|
||||
_images_path = images_path
|
||||
|
||||
|
||||
def image(name):
|
||||
if name in _IMAGES:
|
||||
return _IMAGES[name]
|
||||
|
||||
if _images_path:
|
||||
path = _os_path.join(_images_path, name + '.png')
|
||||
if _os_path.isfile(path):
|
||||
_IMAGES[name] = path
|
||||
return path
|
||||
else:
|
||||
_IMAGES[name] = None
|
||||
|
|
@ -0,0 +1,42 @@
|
|||
#
|
||||
#
|
||||
#
|
||||
|
||||
from gi.repository import Gtk
|
||||
|
||||
|
||||
def _show_icon_menu(icon, button, time, menu):
|
||||
menu.popup(None, None, icon.position_menu, icon, button, time)
|
||||
|
||||
|
||||
def create(app_icon, title, menu_actions, click_action=None):
|
||||
icon = Gtk.StatusIcon.new_from_file(app_icon)
|
||||
icon.set_title(title)
|
||||
icon.set_name(title)
|
||||
|
||||
if click_action:
|
||||
if type(click_action) == tuple:
|
||||
function = click_action[0]
|
||||
args = click_action[1:]
|
||||
icon.connect('activate', function, *args)
|
||||
else:
|
||||
icon.connect('activate', click_action)
|
||||
|
||||
if menu_actions:
|
||||
if type(menu_actions) == list:
|
||||
menu = Gtk.Menu()
|
||||
for action in menu_actions:
|
||||
if action:
|
||||
item = Gtk.MenuItem(action[0])
|
||||
function = action[1]
|
||||
args = action[2:] if len(action) > 2 else ()
|
||||
item.connect('activate', function, *args)
|
||||
menu.append(item)
|
||||
else:
|
||||
menu.append(Gtk.SeparatorMenuItem())
|
||||
menu.show_all()
|
||||
icon.connect('popup_menu', _show_icon_menu, menu)
|
||||
else:
|
||||
icon.connect('popup_menu', menu_actions)
|
||||
|
||||
return icon
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
#
|
||||
# Optional desktop notifications.
|
||||
#
|
||||
|
||||
try:
|
||||
import notify2 as _notify
|
||||
|
||||
|
||||
available = True
|
||||
_app_title = None
|
||||
_images = lambda x: None
|
||||
_notifications = {}
|
||||
|
||||
|
||||
def start(app_title, images=None):
|
||||
global _app_title, _images
|
||||
_notify.init(app_title)
|
||||
_app_title = app_title
|
||||
_images = images or (lambda x: None)
|
||||
|
||||
|
||||
def stop():
|
||||
global _app_title
|
||||
_app_title = None
|
||||
all(n.close() for n in list(_notifications.values()))
|
||||
_notify.uninit()
|
||||
_notifications.clear()
|
||||
|
||||
|
||||
def show(status, title, text, icon=None):
|
||||
if not _app_title:
|
||||
return
|
||||
|
||||
if title in _notifications:
|
||||
notification = _notifications[title]
|
||||
else:
|
||||
_notifications[title] = notification = _notify.Notification(title)
|
||||
|
||||
if text == notification.message:
|
||||
# there's no need to show the same notification twice in a row
|
||||
return
|
||||
|
||||
path = _images('devices/' + title if icon is None else icon)
|
||||
icon = ('error' if status < 0 else 'info') if path is None else path
|
||||
|
||||
notification.update(title, text, icon)
|
||||
notification.show()
|
||||
|
||||
except ImportError:
|
||||
import logging
|
||||
logging.exception("ouch")
|
||||
logging.warn("python-notify2 not found, desktop notifications are disabled")
|
||||
available = False
|
||||
def start(app_title): pass
|
||||
def stop(): pass
|
||||
def show(status, title, text, icon=None): pass
|
||||
|
|
@ -0,0 +1,213 @@
|
|||
#
|
||||
#
|
||||
#
|
||||
|
||||
from gi.repository import Gtk
|
||||
from gi.repository import Gdk
|
||||
from gi.repository import GdkPixbuf
|
||||
|
||||
from .. import constants as C
|
||||
|
||||
|
||||
_DEVICE_ICON_SIZE = 48
|
||||
_STATUS_ICON_SIZE = 64
|
||||
_PLACEHOLDER = '~'
|
||||
_images = None
|
||||
_MAX_DEVICES = 7
|
||||
|
||||
_ICONS = {}
|
||||
|
||||
|
||||
def _icon(icon, title, size=_DEVICE_ICON_SIZE, fallback=None):
|
||||
icon = icon or Gtk.Image()
|
||||
|
||||
if title and title in _ICONS:
|
||||
icon.set_from_pixbuf(_ICONS[title])
|
||||
else:
|
||||
icon_file = _images(title) if title else None
|
||||
if icon_file:
|
||||
pixbuf = GdkPixbuf.Pixbuf().new_from_file(icon_file)
|
||||
if pixbuf.get_width() > size or pixbuf.get_height() > size:
|
||||
if pixbuf.get_width() > pixbuf.get_height():
|
||||
new_width = size
|
||||
new_height = size * pixbuf.get_height() / pixbuf.get_width()
|
||||
else:
|
||||
new_width = size * pixbuf.get_width() / pixbuf.get_height()
|
||||
new_height = size
|
||||
pixbuf = pixbuf.scale_simple(new_width, new_height, GdkPixbuf.InterpType.HYPER)
|
||||
icon.set_from_pixbuf(pixbuf)
|
||||
_ICONS[title] = pixbuf
|
||||
elif fallback:
|
||||
icon.set_from_icon_name(fallback, size if size < _DEVICE_ICON_SIZE else Gtk.IconSize.DIALOG)
|
||||
|
||||
if size >= _DEVICE_ICON_SIZE:
|
||||
icon.set_size_request(size, size)
|
||||
return icon
|
||||
|
||||
|
||||
def update(window, ur_available, devices):
|
||||
if not window or not window.get_child():
|
||||
return
|
||||
controls = list(window.get_child().get_children())
|
||||
|
||||
first = controls[0]
|
||||
first.set_visible(not ur_available or not devices)
|
||||
if ur_available:
|
||||
ur_status = C.FOUND_RECEIVER if devices else C.NO_DEVICES
|
||||
else:
|
||||
ur_status = C.NO_RECEIVER
|
||||
_, label = first.get_children()
|
||||
label.set_markup('<big><b>%s</b></big>\n%s' % (C.UNIFYING_RECEIVER, ur_status))
|
||||
|
||||
for index in range(1, _MAX_DEVICES):
|
||||
box = controls[index]
|
||||
devstatus = [d for d in devices if d.number == index]
|
||||
devstatus = devstatus[0] if devstatus else None
|
||||
box.set_visible(devstatus is not None)
|
||||
|
||||
if devstatus:
|
||||
box.set_sensitive(devstatus.code >= 0)
|
||||
icon, expander = box.get_children()
|
||||
if not expander.get_data('devstatus'):
|
||||
expander.set_data('devstatus', devstatus,)
|
||||
_icon(icon, 'devices/' + devstatus.name, fallback=devstatus.type.lower())
|
||||
|
||||
label = expander.get_label_widget()
|
||||
if expander.get_expanded():
|
||||
label.set_markup('<big><b>%s</b></big>' % devstatus.name)
|
||||
else:
|
||||
label.set_markup('<big><b>%s</b></big>\n%s' % (devstatus.name, devstatus.props['text']))
|
||||
|
||||
ebox = expander.get_child()
|
||||
|
||||
# refresh_button = ebox.get_children()[0]
|
||||
# refresh_button.connect('activate', devstatus.refresh)
|
||||
|
||||
texts = []
|
||||
|
||||
battery_icon = ebox.get_children()[-1]
|
||||
if 'battery-level' in devstatus.props:
|
||||
level = devstatus.props['battery-level']
|
||||
icon_name = 'battery/' + str((level + 10) // 20)
|
||||
_icon(battery_icon, icon_name, _STATUS_ICON_SIZE)
|
||||
texts.append('Battery: ' + str(level) + '%')
|
||||
else:
|
||||
_icon(battery_icon, 'battery/unknown', _STATUS_ICON_SIZE)
|
||||
texts.append('Battery: unknown')
|
||||
battery_icon.set_tooltip_text(texts[-1])
|
||||
|
||||
light_icon = ebox.get_children()[-2]
|
||||
if 'light-level' in devstatus.props:
|
||||
lux = devstatus.props['light-level']
|
||||
icon_name = 'light/' + str((lux + 50) // 100)
|
||||
_icon(light_icon, icon_name, _STATUS_ICON_SIZE)
|
||||
|
||||
texts.append('Light: ' + str(lux) + ' lux')
|
||||
light_icon.set_tooltip_text(texts[-1])
|
||||
light_icon.set_visible(True)
|
||||
else:
|
||||
light_icon.set_visible(False)
|
||||
|
||||
label = ebox.get_children()[-3]
|
||||
label.set_text('\n'.join(texts))
|
||||
|
||||
def _expander_activate(expander):
|
||||
devstatus = expander.get_data('devstatus')
|
||||
if devstatus:
|
||||
label = expander.get_label_widget()
|
||||
if expander.get_expanded():
|
||||
label.set_markup('<big><b>%s</b></big>\n%s' % (devstatus.name, devstatus.props['text']))
|
||||
else:
|
||||
label.set_markup('<big><b>%s</b></big>' % devstatus.name)
|
||||
|
||||
|
||||
def _device_box(title):
|
||||
icon = _icon(None, 'devices/' + title)
|
||||
icon.set_alignment(0.5, 0)
|
||||
|
||||
label = Gtk.Label()
|
||||
label.set_markup('<big><b>%s</b></big>' % title)
|
||||
label.set_alignment(0, 0.5)
|
||||
label.set_can_focus(False)
|
||||
|
||||
box = Gtk.HBox(spacing=10)
|
||||
box.pack_start(icon, False, False, 0)
|
||||
|
||||
if title == C.UNIFYING_RECEIVER:
|
||||
box.add(label)
|
||||
else:
|
||||
expander = Gtk.Expander()
|
||||
expander.set_can_focus(False)
|
||||
expander.set_label_widget(label)
|
||||
expander.connect('activate', _expander_activate)
|
||||
|
||||
ebox = Gtk.HBox(False, 10)
|
||||
ebox.set_border_width(4)
|
||||
|
||||
# refresh_button = Gtk.Button()
|
||||
# refresh_button.set_image(_icon(None, None, size=Gtk.IconSize.SMALL_TOOLBAR, fallback='reload'))
|
||||
# refresh_button.set_focus_on_click(False)
|
||||
# refresh_button.set_can_focus(False)
|
||||
# refresh_button.set_image_position(Gtk.PositionType.TOP)
|
||||
# refresh_button.set_alignment(0.5, 0.5)
|
||||
# refresh_button.set_relief(Gtk.ReliefStyle.NONE)
|
||||
# refresh_button.set_size_request(20, 20)
|
||||
# refresh_button.set_tooltip_text('Refresh')
|
||||
# ebox.pack_start(refresh_button, False, False, 2)
|
||||
|
||||
label = Gtk.Label()
|
||||
label.set_alignment(0, 0.5)
|
||||
ebox.pack_start(label, False, True, 8)
|
||||
|
||||
light_icon = _icon(None, 'light/unknown', _STATUS_ICON_SIZE)
|
||||
ebox.pack_end(light_icon, False, True, 0)
|
||||
|
||||
battery_icon = _icon(None, 'battery/unknown', _STATUS_ICON_SIZE)
|
||||
ebox.pack_end(battery_icon, False, True, 0)
|
||||
|
||||
expander.add(ebox)
|
||||
box.pack_start(expander, True, True, 1)
|
||||
|
||||
box.show_all()
|
||||
box.set_visible(title != _PLACEHOLDER)
|
||||
return box
|
||||
|
||||
|
||||
def create(title, images=None):
|
||||
global _images
|
||||
_images = images or (lambda x: None)
|
||||
|
||||
vbox = Gtk.VBox(spacing=8)
|
||||
vbox.set_border_width(6)
|
||||
|
||||
vbox.add(_device_box(C.UNIFYING_RECEIVER))
|
||||
for i in range(1, _MAX_DEVICES):
|
||||
vbox.add(_device_box(_PLACEHOLDER))
|
||||
vbox.set_visible(True)
|
||||
|
||||
window = Gtk.Window() # Gtk.WindowType.POPUP)
|
||||
window.set_title(title)
|
||||
window.set_icon_from_file(_images('icon'))
|
||||
window.set_keep_above(True)
|
||||
window.set_decorated(False)
|
||||
window.set_skip_taskbar_hint(True)
|
||||
window.set_skip_pager_hint(True)
|
||||
window.set_deletable(False)
|
||||
window.set_resizable(False)
|
||||
window.set_position(Gtk.WindowPosition.MOUSE)
|
||||
window.set_type_hint(Gdk.WindowTypeHint.TOOLTIP)
|
||||
window.connect('focus-out-event', _hide)
|
||||
|
||||
window.add(vbox)
|
||||
return window
|
||||
|
||||
|
||||
def _hide(window, _):
|
||||
window.set_visible(False)
|
||||
|
||||
|
||||
def toggle(_, window):
|
||||
if window.get_visible():
|
||||
window.set_visible(False)
|
||||
else:
|
||||
window.present()
|
||||
|
|
@ -0,0 +1,200 @@
|
|||
#
|
||||
#
|
||||
#
|
||||
|
||||
import logging
|
||||
import threading
|
||||
import time
|
||||
|
||||
import constants as C
|
||||
from logitech.unifying_receiver import api
|
||||
from logitech.unifying_receiver.listener import EventsListener
|
||||
from logitech import devices
|
||||
|
||||
|
||||
_STATUS_TIMEOUT = 97 # seconds
|
||||
_THREAD_SLEEP = 7 # seconds
|
||||
_FORGET_TIMEOUT = 5 * 60 # seconds
|
||||
|
||||
|
||||
class _DevStatus(api.AttachedDeviceInfo):
|
||||
timestamp = time.time()
|
||||
code = devices.STATUS.CONNECTED
|
||||
props = {devices.PROPS.TEXT: devices.STATUS_NAME[devices.STATUS.CONNECTED]}
|
||||
refresh = None
|
||||
|
||||
|
||||
class WatcherThread(threading.Thread):
|
||||
def __init__(self, notify_callback=None):
|
||||
super(WatcherThread, self).__init__(name='WatcherThread')
|
||||
self.daemon = True
|
||||
self.active = False
|
||||
|
||||
self.notify = notify_callback
|
||||
self.status_text = None
|
||||
self.status_changed = threading.Event()
|
||||
|
||||
self.listener = None
|
||||
self.devices = {}
|
||||
|
||||
def run(self):
|
||||
self.active = True
|
||||
self._notify(0, C.UNIFYING_RECEIVER, C.SCANNING)
|
||||
|
||||
while self.active:
|
||||
if self.listener is None:
|
||||
receiver = api.open()
|
||||
if receiver:
|
||||
self._notify(1, C.UNIFYING_RECEIVER, C.FOUND_RECEIVER)
|
||||
for devinfo in api.list_devices(receiver):
|
||||
devstatus = _DevStatus(*devinfo)
|
||||
self.devices[devinfo.number] = devstatus
|
||||
self._notify(devices.STATUS.CONNECTED, devstatus.name, devices.STATUS_NAME[devices.STATUS.CONNECTED])
|
||||
self.listener = EventsListener(receiver, self._events_callback)
|
||||
self.listener.start()
|
||||
self._update_status()
|
||||
else:
|
||||
self._notify(-1, C.UNIFYING_RECEIVER, C.NO_RECEIVER)
|
||||
elif not self.listener.active:
|
||||
self.listener = None
|
||||
self._notify(-1, C.UNIFYING_RECEIVER, C.NO_RECEIVER)
|
||||
self.devices.clear()
|
||||
|
||||
if self.active:
|
||||
update_icon = True
|
||||
if self.listener and self.devices:
|
||||
update_icon &= self._check_old_statuses()
|
||||
|
||||
if self.active:
|
||||
if update_icon:
|
||||
self._update_status()
|
||||
time.sleep(_THREAD_SLEEP)
|
||||
|
||||
def stop(self):
|
||||
self.active = False
|
||||
if self.listener:
|
||||
self.listener.stop()
|
||||
api.close(self.listener.receiver)
|
||||
|
||||
def has_receiver(self):
|
||||
return self.listener is not None and self.listener.active
|
||||
|
||||
def request_all_statuses(self, _=None):
|
||||
updated = False
|
||||
|
||||
for d in range(1, 7):
|
||||
devstatus = self.devices.get(d)
|
||||
if devstatus:
|
||||
status = devices.request_status(devstatus, self.listener)
|
||||
updated |= self._device_status_changed(devstatus, status)
|
||||
else:
|
||||
devstatus = self._new_device(d)
|
||||
updated |= devstatus is not None
|
||||
|
||||
if updated:
|
||||
self._update_status()
|
||||
|
||||
def _check_old_statuses(self):
|
||||
updated = False
|
||||
|
||||
for devstatus in list(self.devices.values()):
|
||||
if time.time() - devstatus.timestamp > _STATUS_TIMEOUT:
|
||||
status = devices.ping(devstatus, self.listener)
|
||||
updated |= self._device_status_changed(devstatus, status)
|
||||
|
||||
return updated
|
||||
|
||||
def _new_device(self, device):
|
||||
devinfo = api.get_device_info(self.listener.receiver, device)
|
||||
if devinfo:
|
||||
devstatus = _DevStatus(*devinfo)
|
||||
self.devices[device] = devstatus
|
||||
self._notify(devstatus.code, devstatus.name, devstatus.props[devices.PROPS.TEXT])
|
||||
return devinfo
|
||||
|
||||
def _events_callback(self, code, device, data):
|
||||
logging.debug("%s: event %02x %d %s", time.asctime(), code, device, repr(data))
|
||||
|
||||
updated = False
|
||||
|
||||
if device in self.devices:
|
||||
devstatus = self.devices[device]
|
||||
if code == 0x10 and data[0] == 'b\x8F':
|
||||
updated = True
|
||||
self._device_status_changed(devstatus, devices.STATUS.UNAVAILABLE)
|
||||
elif code == 0x11:
|
||||
status = devices.process_event(devstatus, self.listener, data)
|
||||
updated |= self._device_status_changed(devstatus, status)
|
||||
else:
|
||||
logging.warn("unknown event code %02x", code)
|
||||
elif device:
|
||||
logging.debug("got event (%d, %d, %s) for new device", code, device, repr(data))
|
||||
self._new_device(device)
|
||||
updated = True
|
||||
else:
|
||||
logging.warn("don't know how to handle event (%d, %d, %s)", code, device, data)
|
||||
|
||||
if updated:
|
||||
self._update_status()
|
||||
|
||||
def _device_status_changed(self, devstatus, status):
|
||||
if status is None:
|
||||
return False
|
||||
|
||||
old_status_code = devstatus.code
|
||||
devstatus.timestamp = time.time()
|
||||
|
||||
if type(status) == int:
|
||||
devstatus.code = status
|
||||
if devstatus.code in devices.STATUS_NAME:
|
||||
devstatus.props[devices.PROPS.TEXT] = devices.STATUS_NAME[devstatus.code]
|
||||
else:
|
||||
devstatus.code = status[0]
|
||||
devstatus.props.update(status[1])
|
||||
|
||||
if old_status_code != devstatus.code:
|
||||
logging.debug("%s: device status changed %s => %s: %s", time.asctime(), old_status_code, devstatus.code, devstatus.props)
|
||||
# if not (devstatus.code == 0 and old_status_code > 0):
|
||||
self._notify(devstatus.code, devstatus.name, devstatus.props[devices.PROPS.TEXT])
|
||||
|
||||
return True
|
||||
|
||||
def _notify(self, *args):
|
||||
if self.notify:
|
||||
self.notify(*args)
|
||||
|
||||
def notify_full(self):
|
||||
if self.listener and self.listener.active:
|
||||
if self.devices:
|
||||
for devstatus in self.devices.values():
|
||||
self._notify(0, devstatus.name, devstatus.props[devices.PROPS.TEXT])
|
||||
else:
|
||||
self._notify(0, C.UNIFYING_RECEIVER, C.NO_DEVICES)
|
||||
else:
|
||||
self._notify(-1, C.UNIFYING_RECEIVER, C.NO_RECEIVER)
|
||||
|
||||
def _update_status(self):
|
||||
last_status_text = self.status_text
|
||||
|
||||
if self.listener and self.listener.active:
|
||||
if self.devices:
|
||||
all_statuses = []
|
||||
for d in self.devices:
|
||||
devstatus = self.devices[d]
|
||||
status_text = devstatus.props[devices.PROPS.TEXT]
|
||||
if status_text:
|
||||
if ' ' in status_text:
|
||||
all_statuses.append(devstatus.name)
|
||||
all_statuses.append(' ' + status_text)
|
||||
else:
|
||||
all_statuses.append(devstatus.name + ' ' + status_text)
|
||||
else:
|
||||
all_statuses.append(devstatus.name)
|
||||
self.status_text = '\n'.join(all_statuses)
|
||||
else:
|
||||
self.status_text = C.NO_DEVICES
|
||||
else:
|
||||
self.status_text = C.NO_RECEIVER
|
||||
|
||||
if self.status_text != last_status_text:
|
||||
self.status_changed.set()
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 20 KiB |
Binary file not shown.
|
Before Width: | Height: | Size: 109 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 17 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 17 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 18 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 19 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 21 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 20 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 26 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 2.9 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 3.2 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 7.3 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 40 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 14 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 20 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 18 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 21 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 3.2 KiB |
|
|
@ -35,19 +35,8 @@ def scan_devices(receiver):
|
|||
if keys is not None and keys:
|
||||
print " %d reprogrammable keys found" % len(keys)
|
||||
for k in keys:
|
||||
flags = ''
|
||||
if k.flags & KEY_FLAG.REPROGRAMMABLE:
|
||||
flags += ' reprogrammable'
|
||||
if k.flags & KEY_FLAG.FN_SENSITIVE:
|
||||
flags += ' fn-sensitive'
|
||||
if k.flags & KEY_FLAG.NONSTANDARD:
|
||||
flags += ' nonstandard'
|
||||
if k.flags & KEY_FLAG.IS_FN:
|
||||
flags += ' is-fn'
|
||||
if k.flags & KEY_FLAG.MSE:
|
||||
flags += ' mse'
|
||||
|
||||
print " %2d: %s => %s :%s" % (k.index, KEY_NAME[k.id], KEY_NAME[k.task], flags)
|
||||
flags = ','.join(KEY_FLAG_NAME[f] for f in KEY_FLAG_NAME if k.flags & f)
|
||||
print " %2d: %-12s => %-12s :%s" % (k.index, KEY_NAME[k.id], KEY_NAME[k.task], flags)
|
||||
|
||||
print "--------"
|
||||
|
||||
|
|
|
|||
|
|
@ -5,20 +5,55 @@
|
|||
from . import k750
|
||||
from .constants import *
|
||||
|
||||
from ..unifying_receiver import api as _api
|
||||
from ..unifying_receiver.common import FallbackDict as _FDict
|
||||
|
||||
_REQUEST_STATUS_FUNCTIONS = {
|
||||
k750.NAME : k750.request_status,
|
||||
}
|
||||
|
||||
def ping(devinfo, listener):
|
||||
reply = listener.request(_api.ping, devinfo.number)
|
||||
return STATUS.CONNECTED if reply else STATUS.UNAVAILABLE
|
||||
|
||||
|
||||
def default_request_status(devinfo, listener):
|
||||
if _api.C.FEATURE.BATTERY in devinfo.features:
|
||||
reply = listener.request(_api.get_device_battery_level, devinfo.number, features_array=devinfo.features)
|
||||
if reply:
|
||||
discharge, dischargeNext, status = reply
|
||||
return STATUS.CONNECTED, {PROPS.BATTERY_LEVEL: discharge}
|
||||
|
||||
|
||||
def default_process_event(devinfo, listener, data):
|
||||
feature_index = ord(data[0:1])
|
||||
feature = devinfo.features[feature_index]
|
||||
|
||||
if feature == _api.C.FEATURE.BATTERY:
|
||||
if ord(data[1:2]) & 0xF0 == 0:
|
||||
discharge = ord(data[2:3])
|
||||
status = _api.C.BATTERY_STATUS[ord(data[3:4])]
|
||||
return STATUS.CONNECTED, {BATTERY_LEVEL: discharge, BATTERY_STATUS: status}
|
||||
# ?
|
||||
elif feature == _api.C.FEATURE.REPROGRAMMABLE_KEYS:
|
||||
if ord(data[1:2]) & 0xF0 == 0:
|
||||
print 'reprogrammable key', repr(data)
|
||||
# TODO
|
||||
pass
|
||||
# ?
|
||||
elif feature == _api.C.FEATURE.WIRELESS:
|
||||
if ord(data[1:2]) & 0xF0 == 0:
|
||||
# TODO
|
||||
pass
|
||||
# ?
|
||||
|
||||
|
||||
_REQUEST_STATUS_FUNCTIONS = _FDict()
|
||||
_REQUEST_STATUS_FUNCTIONS[k750.NAME] = k750.request_status
|
||||
|
||||
def request_status(devinfo, listener):
|
||||
if devinfo.name in _REQUEST_STATUS_FUNCTIONS:
|
||||
return _REQUEST_STATUS_FUNCTIONS[devinfo.name](devinfo, listener)
|
||||
return _REQUEST_STATUS_FUNCTIONS[devinfo.name](devinfo, listener) or default_request_status(devinfo, listener) or ping(devinfo, listener)
|
||||
|
||||
|
||||
_PROCESS_EVENT_FUNCTIONS = {
|
||||
k750.NAME : k750.process_event,
|
||||
}
|
||||
_PROCESS_EVENT_FUNCTIONS = _FDict()
|
||||
_PROCESS_EVENT_FUNCTIONS[k750.NAME] = k750.process_event
|
||||
|
||||
def process_event(devinfo, listener, data):
|
||||
if devinfo.name in _PROCESS_EVENT_FUNCTIONS:
|
||||
return _PROCESS_EVENT_FUNCTIONS[devinfo.name](devinfo, listener, data)
|
||||
return default_process_event(devinfo, listener, data) or _PROCESS_EVENT_FUNCTIONS[devinfo.name](devinfo, listener, data)
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@
|
|||
#
|
||||
#
|
||||
|
||||
DEVICE_STATUS = type('DEVICE_STATUS', (),
|
||||
STATUS = type('STATUS', (),
|
||||
dict(
|
||||
UNKNOWN=None,
|
||||
UNAVAILABLE=-1,
|
||||
|
|
@ -10,12 +10,21 @@ DEVICE_STATUS = type('DEVICE_STATUS', (),
|
|||
# ACTIVE=1,
|
||||
))
|
||||
|
||||
PROPS = type('PROPS', (),
|
||||
dict(
|
||||
TEXT='text',
|
||||
BATTERY_LEVEL='battery-level',
|
||||
BATTERY_STATUS='battery-status',
|
||||
LIGHT_LUX='lux',
|
||||
LIGHT_LEVEL='light-level',
|
||||
))
|
||||
|
||||
|
||||
from collections import defaultdict
|
||||
|
||||
DEVICE_STATUS_NAME = defaultdict(lambda x: None)
|
||||
DEVICE_STATUS_NAME[DEVICE_STATUS.UNAVAILABLE] = 'disconnected'
|
||||
DEVICE_STATUS_NAME[DEVICE_STATUS.CONNECTED] = 'connected'
|
||||
# DEVICE_STATUS_NAME[DEVICE_STATUS.ACTIVE] = 'active'
|
||||
STATUS_NAME = defaultdict(lambda x: None)
|
||||
STATUS_NAME[STATUS.UNAVAILABLE] = 'disconnected'
|
||||
STATUS_NAME[STATUS.CONNECTED] = 'connected'
|
||||
# STATUS_NAME[STATUS.ACTIVE] = 'active'
|
||||
|
||||
del defaultdict
|
||||
|
|
|
|||
|
|
@ -3,10 +3,10 @@
|
|||
#
|
||||
|
||||
import logging
|
||||
import struct
|
||||
from struct import unpack as _unpack
|
||||
|
||||
from ..unifying_receiver import api as _api
|
||||
from .constants import *
|
||||
from . import constants as C
|
||||
|
||||
#
|
||||
#
|
||||
|
|
@ -14,10 +14,7 @@ from .constants import *
|
|||
|
||||
NAME = 'Wireless Solar Keyboard K750'
|
||||
|
||||
_STATUS_NAMES = ('excellent', 'good', 'okay', 'poor', 'very low')
|
||||
|
||||
_CHARGE_LIMITS = (75, 40, 20, 10, -1)
|
||||
_LIGHTING_LIMITS = (400, 200, 50, 20, -1)
|
||||
|
||||
#
|
||||
#
|
||||
|
|
@ -25,58 +22,55 @@ _LIGHTING_LIMITS = (400, 200, 50, 20, -1)
|
|||
|
||||
def _trigger_solar_charge_events(receiver, devinfo):
|
||||
return _api.request(receiver, devinfo.number,
|
||||
feature=_api.FEATURE.SOLAR_CHARGE, function=b'\x03', params=b'\x78\x01',
|
||||
feature=_api.C.FEATURE.SOLAR_CHARGE, function=b'\x03', params=b'\x78\x01',
|
||||
features_array=devinfo.features)
|
||||
|
||||
|
||||
def _charge_status(data):
|
||||
charge, lux = struct.unpack('!BH', data[2:5])
|
||||
charge, lux = _unpack('!BH', data[2:5])
|
||||
|
||||
d = {}
|
||||
|
||||
for i in range(0, len(_CHARGE_LIMITS)):
|
||||
if charge >= _CHARGE_LIMITS[i]:
|
||||
charge_index = i
|
||||
break
|
||||
text = 'Charge %d%% (%s)' % (charge, _STATUS_NAMES[charge_index])
|
||||
else:
|
||||
charge_index = 0
|
||||
d[C.PROPS.BATTERY_LEVEL] = charge
|
||||
text = 'Battery %d%%' % charge
|
||||
|
||||
if lux > 0:
|
||||
for i in range(0, len(_CHARGE_LIMITS)):
|
||||
if lux > _LIGHTING_LIMITS[i]:
|
||||
lighting_index = i
|
||||
break
|
||||
text += ', Lighting %s (%d lux)' % (_STATUS_NAMES[lighting_index], lux)
|
||||
d[C.PROPS.LIGHT_LEVEL] = lux
|
||||
text = 'Light: %d lux' % lux + ', ' + text
|
||||
|
||||
return 0x10 << charge_index, text
|
||||
d[C.PROPS.TEXT] = text
|
||||
return 0x10 << charge_index, d
|
||||
|
||||
|
||||
def request_status(devinfo, listener):
|
||||
# Constantly requesting the solar charge status triggers a flood of events,
|
||||
# which appear to drain the battery rather fast.
|
||||
# Instead, ping the device for on/off status, and only ask for solar charge
|
||||
# status when the user presses the solar key on the keyboard.
|
||||
#
|
||||
# reply = listener.request(_trigger_solar_charge_events, devinfo)
|
||||
# if reply is None:
|
||||
# return DEVICE_STATUS.UNAVAILABLE
|
||||
|
||||
reply = listener.request(_api.ping, devinfo.number)
|
||||
if not reply:
|
||||
return DEVICE_STATUS.UNAVAILABLE
|
||||
reply = listener.request(_trigger_solar_charge_events, devinfo)
|
||||
if reply is None:
|
||||
return C.STATUS.UNAVAILABLE
|
||||
|
||||
|
||||
def process_event(devinfo, listener, data):
|
||||
if data[:2] == b'\x09\x00' and data[7:11] == b'GOOD':
|
||||
# usually sent after the keyboard is turned on
|
||||
return _charge_status(data)
|
||||
|
||||
if data[:2] == b'\x09\x10' and data[7:11] == b'GOOD':
|
||||
# regular solar charge events
|
||||
return _charge_status(data)
|
||||
|
||||
if data[:2] == b'\x09\x20' and data[7:11] == b'GOOD':
|
||||
logging.debug("Solar key pressed")
|
||||
if _trigger_solar_charge_events(listener.receiver, devinfo) is None:
|
||||
return C.STATUS.UNAVAILABLE
|
||||
return _charge_status(data)
|
||||
|
||||
if data[:2] == b'\x05\x00':
|
||||
# wireless device status
|
||||
if data[2:5] == b'\x01\x01\x01':
|
||||
logging.debug("Keyboard just started")
|
||||
return DEVICE_STATUS.CONNECTED
|
||||
elif data[:2] == b'\x09\x00' and data[7:11] == b'GOOD':
|
||||
# usually sent after the keyboard is turned on
|
||||
return _charge_status(data)
|
||||
elif data[:2] == b'\x09\x10' and data[7:11] == b'GOOD':
|
||||
# regular solar charge events
|
||||
return _charge_status(data)
|
||||
elif data[:2] == b'\x09\x20' and data[7:11] == b'GOOD':
|
||||
logging.debug("Solar key pressed")
|
||||
if _trigger_solar_charge_events(listener.receiver, devinfo) is None:
|
||||
return DEVICE_STATUS.UNAVAILABLE
|
||||
return _charge_status(data)
|
||||
return C.STATUS.CONNECTED
|
||||
|
|
|
|||
|
|
@ -3,14 +3,17 @@
|
|||
#
|
||||
|
||||
import logging
|
||||
import struct
|
||||
from binascii import hexlify
|
||||
from struct import pack as _pack
|
||||
from struct import unpack as _unpack
|
||||
from binascii import hexlify as _hexlify
|
||||
|
||||
from .common import *
|
||||
from .constants import *
|
||||
from .exceptions import *
|
||||
from . import base
|
||||
from .unhandled import _publish as _unhandled_publish
|
||||
from .common import FirmwareInfo
|
||||
from .common import AttachedDeviceInfo
|
||||
from .common import ReprogrammableKeyInfo
|
||||
from . import constants as C
|
||||
from . import exceptions as E
|
||||
from . import unhandled as _unhandled
|
||||
from . import base as _base
|
||||
|
||||
|
||||
_LOG_LEVEL = 5
|
||||
|
|
@ -25,10 +28,10 @@ def open():
|
|||
|
||||
:returns: An open file handle for the found receiver, or ``None``.
|
||||
"""
|
||||
for rawdevice in base.list_receiver_devices():
|
||||
for rawdevice in _base.list_receiver_devices():
|
||||
_l.log(_LOG_LEVEL, "checking %s", rawdevice)
|
||||
|
||||
receiver = base.try_open(rawdevice.path)
|
||||
receiver = _base.try_open(rawdevice.path)
|
||||
if receiver:
|
||||
return receiver
|
||||
|
||||
|
|
@ -36,7 +39,7 @@ def open():
|
|||
|
||||
|
||||
"""Closes a HID device handle."""
|
||||
close = base.close
|
||||
close = _base.close
|
||||
|
||||
|
||||
def request(handle, device, feature, function=b'\x00', params=b'', features_array=None):
|
||||
|
|
@ -58,7 +61,7 @@ def request(handle, device, feature, function=b'\x00', params=b'', features_arra
|
|||
"""
|
||||
|
||||
feature_index = None
|
||||
if feature == FEATURE.ROOT:
|
||||
if feature == C.FEATURE.ROOT:
|
||||
feature_index = b'\x00'
|
||||
else:
|
||||
if features_array is None:
|
||||
|
|
@ -67,13 +70,13 @@ def request(handle, device, feature, function=b'\x00', params=b'', features_arra
|
|||
_l.log(_LOG_LEVEL, "(%d,%d) no features array available", handle, device)
|
||||
return None
|
||||
if feature in features_array:
|
||||
feature_index = struct.pack('!B', features_array.index(feature))
|
||||
feature_index = _pack('!B', features_array.index(feature))
|
||||
|
||||
if feature_index is None:
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported", handle, device, hexlify(feature), FEATURE_NAME[feature])
|
||||
raise FeatureNotSupported(device, feature)
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported", handle, device, _hexlify(feature), C.FEATURE_NAME[feature])
|
||||
raise E.FeatureNotSupported(device, feature)
|
||||
|
||||
return base.request(handle, device, feature_index + function, params)
|
||||
return _base.request(handle, device, feature_index + function, params)
|
||||
|
||||
|
||||
def ping(handle, device):
|
||||
|
|
@ -93,13 +96,13 @@ def ping(handle, device):
|
|||
|
||||
if reply_device != device:
|
||||
# oops
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: reply for another device %d: %s", handle, device, reply_device, hexlify(reply_data))
|
||||
_unhandled_publish(reply_code, reply_device, reply_data)
|
||||
return _status(base.read(handle))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: reply for another device %d: %s", handle, device, reply_device, _hexlify(reply_data))
|
||||
_unhandled._publish(reply_code, reply_device, reply_data)
|
||||
return _status(_base.read(handle))
|
||||
|
||||
if (reply_code == 0x11 and reply_data[:2] == b'\x00\x10' and reply_data[4:5] == ping_marker):
|
||||
# ping ok
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: ok [%s]", handle, device, hexlify(reply_data))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: ok [%s]", handle, device, _hexlify(reply_data))
|
||||
return True
|
||||
|
||||
if (reply_code == 0x10 and reply_data[:2] == b'\x8F\x00'):
|
||||
|
|
@ -111,19 +114,19 @@ def ping(handle, device):
|
|||
# some devices may reply with a SOLAR_CHARGE event before the
|
||||
# ping_ok reply, especially right after the device connected to the
|
||||
# receiver
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: solar status %s", handle, device, hexlify(reply_data))
|
||||
_unhandled_publish(reply_code, reply_device, reply_data)
|
||||
return _status(base.read(handle))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: solar status [%s]", handle, device, _hexlify(reply_data))
|
||||
_unhandled._publish(reply_code, reply_device, reply_data)
|
||||
return _status(_base.read(handle))
|
||||
|
||||
# ugh
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: unknown reply for this device: %d=[%s]", handle, device, reply[0], hexlify(reply[2]))
|
||||
_unhandled_publish(reply_code, reply_device, reply_data)
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) ping: unknown reply for this device: %d=[%s]", handle, device, reply_code, _hexlify(reply_data))
|
||||
_unhandled._publish(reply_code, reply_device, reply_data)
|
||||
return None
|
||||
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) pinging", handle, device)
|
||||
base.write(handle, device, b'\x00\x10\x00\x00' + ping_marker)
|
||||
_base.write(handle, device, b'\x00\x10\x00\x00' + ping_marker)
|
||||
# pings may take a while to reply success
|
||||
return _status(base.read(handle, base.DEFAULT_TIMEOUT * 3))
|
||||
return _status(_base.read(handle, _base.DEFAULT_TIMEOUT * 3))
|
||||
|
||||
|
||||
def find_device_by_name(handle, device_name):
|
||||
|
|
@ -133,7 +136,7 @@ def find_device_by_name(handle, device_name):
|
|||
"""
|
||||
_l.log(_LOG_LEVEL, "(%d,) searching for device '%s'", handle, device_name)
|
||||
|
||||
for device in range(1, 1 + base.MAX_ATTACHED_DEVICES):
|
||||
for device in range(1, 1 + _base.MAX_ATTACHED_DEVICES):
|
||||
features_array = get_device_features(handle, device)
|
||||
if features_array:
|
||||
d_name = get_device_name(handle, device, features_array)
|
||||
|
|
@ -150,7 +153,7 @@ def list_devices(handle):
|
|||
|
||||
devices = []
|
||||
|
||||
for device in range(1, 1 + base.MAX_ATTACHED_DEVICES):
|
||||
for device in range(1, 1 + _base.MAX_ATTACHED_DEVICES):
|
||||
features_array = get_device_features(handle, device)
|
||||
if features_array:
|
||||
devices.append(get_device_info(handle, device, features_array=features_array))
|
||||
|
|
@ -181,31 +184,31 @@ def get_feature_index(handle, device, feature):
|
|||
|
||||
:returns: An int, or ``None`` if the feature is not available.
|
||||
"""
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) get feature index <%s:%s>", handle, device, hexlify(feature), FEATURE_NAME[feature])
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) get feature index <%s:%s>", handle, device, _hexlify(feature), C.FEATURE_NAME[feature])
|
||||
if len(feature) != 2:
|
||||
raise ValueError("invalid feature <%s>: it must be a two-byte string" % feature)
|
||||
|
||||
# FEATURE.ROOT should always be available for any attached devices
|
||||
reply = base.request(handle, device, FEATURE.ROOT, feature)
|
||||
reply = _base.request(handle, device, C.FEATURE.ROOT, feature)
|
||||
if reply:
|
||||
# only consider active and supported features
|
||||
feature_index = ord(reply[0:1])
|
||||
if feature_index:
|
||||
feature_flags = ord(reply[1:2]) & 0xE0
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) feature <%s:%s> has index %d flags %02x", handle, device, hexlify(feature), FEATURE_NAME[feature], feature_index, feature_flags)
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) feature <%s:%s> has index %d flags %02x", handle, device, _hexlify(feature), C.FEATURE_NAME[feature], feature_index, feature_flags)
|
||||
if feature_flags == 0:
|
||||
return feature_index
|
||||
|
||||
if feature_flags & 0x80:
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported: obsolete", handle, device, hexlify(feature), FEATURE_NAME[feature])
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported: obsolete", handle, device, _hexlify(feature), C.FEATURE_NAME[feature])
|
||||
if feature_flags & 0x40:
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported: hidden", handle, device, hexlify(feature), FEATURE_NAME[feature])
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported: hidden", handle, device, _hexlify(feature), C.FEATURE_NAME[feature])
|
||||
if feature_flags & 0x20:
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported: Logitech internal", handle, device, hexlify(feature), FEATURE_NAME[feature])
|
||||
raise FeatureNotSupported(device, feature)
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported: Logitech internal", handle, device, _hexlify(feature), C.FEATURE_NAME[feature])
|
||||
raise E.FeatureNotSupported(device, feature)
|
||||
else:
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported by the device", handle, device, hexlify(feature), FEATURE_NAME[feature])
|
||||
raise FeatureNotSupported(device, feature)
|
||||
_l.warn("(%d,%d) feature <%s:%s> not supported by the device", handle, device, _hexlify(feature), C.FEATURE_NAME[feature])
|
||||
raise E.FeatureNotSupported(device, feature)
|
||||
|
||||
|
||||
def get_device_features(handle, device):
|
||||
|
|
@ -218,7 +221,7 @@ def get_device_features(handle, device):
|
|||
|
||||
# get the index of the FEATURE_SET
|
||||
# FEATURE.ROOT should always be available for all devices
|
||||
fs_index = base.request(handle, device, FEATURE.ROOT, FEATURE.FEATURE_SET)
|
||||
fs_index = _base.request(handle, device, C.FEATURE.ROOT, C.FEATURE.FEATURE_SET)
|
||||
if fs_index is None:
|
||||
# _l.warn("(%d,%d) FEATURE_SET not available", handle, device)
|
||||
return None
|
||||
|
|
@ -228,7 +231,7 @@ def get_device_features(handle, device):
|
|||
# even if unknown.
|
||||
|
||||
# get the number of active features the device has
|
||||
features_count = base.request(handle, device, fs_index + b'\x00')
|
||||
features_count = _base.request(handle, device, fs_index + b'\x00')
|
||||
if not features_count:
|
||||
# this can happen if the device disappeard since the fs_index request
|
||||
# otherwise we should get at least a count of 1 (the FEATURE_SET we've just used above)
|
||||
|
|
@ -238,17 +241,19 @@ def get_device_features(handle, device):
|
|||
features_count = ord(features_count[:1])
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) found %d features", handle, device, features_count)
|
||||
|
||||
# a device may have a maximum of 15 features, other than FEATURE.ROOT
|
||||
features = [None] * 0x10
|
||||
features = [None] * 0x20
|
||||
for index in range(1, 1 + features_count):
|
||||
# for each index, get the feature residing at that index
|
||||
feature = base.request(handle, device, fs_index + b'\x10', struct.pack('!B', index))
|
||||
feature = _base.request(handle, device, fs_index + b'\x10', _pack('!B', index))
|
||||
if feature:
|
||||
feature = feature[0:2].upper()
|
||||
features[index] = feature
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) feature <%s:%s> at index %d", handle, device, hexlify(feature), FEATURE_NAME[feature], index)
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) feature <%s:%s> at index %d", handle, device, _hexlify(feature), C.FEATURE_NAME[feature], index)
|
||||
|
||||
return None if all(c == None for c in features) else features
|
||||
features[0] = C.FEATURE.ROOT
|
||||
while features[-1] is None:
|
||||
del features[-1]
|
||||
return features
|
||||
|
||||
|
||||
def get_device_firmware(handle, device, features_array=None):
|
||||
|
|
@ -259,35 +264,35 @@ def get_device_firmware(handle, device, features_array=None):
|
|||
def _makeFirmwareInfo(level, type, name=None, version=None, build=None, extras=None):
|
||||
return FirmwareInfo(level, type, name, version, build, extras)
|
||||
|
||||
fw_count = request(handle, device, FEATURE.FIRMWARE, features_array=features_array)
|
||||
fw_count = request(handle, device, C.FEATURE.FIRMWARE, features_array=features_array)
|
||||
if fw_count:
|
||||
fw_count = ord(fw_count[:1])
|
||||
|
||||
fw = []
|
||||
for index in range(0, fw_count):
|
||||
index = struct.pack('!B', index)
|
||||
fw_info = request(handle, device, FEATURE.FIRMWARE, function=b'\x10', params=index, features_array=features_array)
|
||||
index = _pack('!B', index)
|
||||
fw_info = request(handle, device, C.FEATURE.FIRMWARE, function=b'\x10', params=index, features_array=features_array)
|
||||
if fw_info:
|
||||
fw_level = ord(fw_info[:1]) & 0x0F
|
||||
if fw_level == 0 or fw_level == 1:
|
||||
fw_type = FIRMWARE_TYPE[fw_level]
|
||||
name, = struct.unpack('!3s', fw_info[1:4])
|
||||
fw_type = C.FIRMWARE_TYPE[fw_level]
|
||||
name, = _unpack('!3s', fw_info[1:4])
|
||||
name = name.decode('ascii')
|
||||
version = ( chr(0x30 + (ord(fw_info[4:5]) >> 4)) +
|
||||
chr(0x30 + (ord(fw_info[4:5]) & 0x0F)) +
|
||||
'.' +
|
||||
chr(0x30 + (ord(fw_info[5:6]) >> 4)) +
|
||||
chr(0x30 + (ord(fw_info[5:6]) & 0x0F)))
|
||||
build, = struct.unpack('!H', fw_info[6:8])
|
||||
build, = _unpack('!H', fw_info[6:8])
|
||||
extras = fw_info[9:].rstrip(b'\x00')
|
||||
if extras:
|
||||
fw_info = _makeFirmwareInfo(level=fw_level, type=fw_type, name=name, version=version, build=build, extras=extras)
|
||||
else:
|
||||
fw_info = _makeFirmwareInfo(level=fw_level, type=fw_type, name=name, version=version, build=build)
|
||||
elif fw_level == 2:
|
||||
fw_info = _makeFirmwareInfo(level=2, type=FIRMWARE_TYPE[2], version=ord(fw_info[1:2]))
|
||||
fw_info = _makeFirmwareInfo(level=2, type=C.FIRMWARE_TYPE[2], version=ord(fw_info[1:2]))
|
||||
else:
|
||||
fw_info = _makeFirmwareInfo(level=fw_level, type=FIRMWARE_TYPE[-1])
|
||||
fw_info = _makeFirmwareInfo(level=fw_level, type=C.FIRMWARE_TYPE[-1])
|
||||
|
||||
fw.append(fw_info)
|
||||
_l.log(_LOG_LEVEL, "(%d:%d) firmware %s", handle, device, fw_info)
|
||||
|
|
@ -301,11 +306,11 @@ def get_device_type(handle, device, features_array=None):
|
|||
:returns: a string describing the device type, or ``None`` if the device is
|
||||
not available or does not support the ``NAME`` feature.
|
||||
"""
|
||||
d_type = request(handle, device, FEATURE.NAME, function=b'\x20', features_array=features_array)
|
||||
d_type = request(handle, device, C.FEATURE.NAME, function=b'\x20', features_array=features_array)
|
||||
if d_type:
|
||||
d_type = ord(d_type[:1])
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) device type %d = %s", handle, device, d_type, DEVICE_TYPE[d_type])
|
||||
return DEVICE_TYPE[d_type]
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) device type %d = %s", handle, device, d_type, C.DEVICE_TYPE[d_type])
|
||||
return C.DEVICE_TYPE[d_type]
|
||||
|
||||
|
||||
def get_device_name(handle, device, features_array=None):
|
||||
|
|
@ -314,14 +319,14 @@ def get_device_name(handle, device, features_array=None):
|
|||
:returns: a string with the device name, or ``None`` if the device is not
|
||||
available or does not support the ``NAME`` feature.
|
||||
"""
|
||||
name_length = request(handle, device, FEATURE.NAME, features_array=features_array)
|
||||
name_length = request(handle, device, C.FEATURE.NAME, features_array=features_array)
|
||||
if name_length:
|
||||
name_length = ord(name_length[:1])
|
||||
|
||||
d_name = b''
|
||||
while len(d_name) < name_length:
|
||||
name_index = struct.pack('!B', len(d_name))
|
||||
name_fragment = request(handle, device, FEATURE.NAME, function=b'\x10', params=name_index, features_array=features_array)
|
||||
name_index = _pack('!B', len(d_name))
|
||||
name_fragment = request(handle, device, C.FEATURE.NAME, function=b'\x10', params=name_index, features_array=features_array)
|
||||
name_fragment = name_fragment[:name_length - len(d_name)]
|
||||
d_name += name_fragment
|
||||
|
||||
|
|
@ -335,24 +340,24 @@ def get_device_battery_level(handle, device, features_array=None):
|
|||
|
||||
:raises FeatureNotSupported: if the device does not support this feature.
|
||||
"""
|
||||
battery = request(handle, device, FEATURE.BATTERY, features_array=features_array)
|
||||
battery = request(handle, device, C.FEATURE.BATTERY, features_array=features_array)
|
||||
if battery:
|
||||
discharge, dischargeNext, status = struct.unpack('!BBB', battery[:3])
|
||||
_l.log(_LOG_LEVEL, "(%d:%d) battery %d%% charged, next level %d%% charge, status %d = %s", discharge, dischargeNext, status, BATTERY_STATUSE[status])
|
||||
return (discharge, dischargeNext, BATTERY_STATUS[status])
|
||||
discharge, dischargeNext, status = _unpack('!BBB', battery[:3])
|
||||
_l.log(_LOG_LEVEL, "(%d:%d) battery %d%% charged, next level %d%% charge, status %d = %s", discharge, dischargeNext, status, C.BATTERY_STATUSE[status])
|
||||
return (discharge, dischargeNext, C.BATTERY_STATUS[status])
|
||||
|
||||
|
||||
def get_device_keys(handle, device, features_array=None):
|
||||
count = request(handle, device, FEATURE.REPROGRAMMABLE_KEYS, features_array=features_array)
|
||||
count = request(handle, device, C.FEATURE.REPROGRAMMABLE_KEYS, features_array=features_array)
|
||||
if count:
|
||||
keys = []
|
||||
|
||||
count = ord(count[:1])
|
||||
for index in range(0, count):
|
||||
keyindex = struct.pack('!B', index)
|
||||
keydata = request(handle, device, FEATURE.REPROGRAMMABLE_KEYS, function=b'\x10', params=keyindex, features_array=features_array)
|
||||
keyindex = _pack('!B', index)
|
||||
keydata = request(handle, device, C.FEATURE.REPROGRAMMABLE_KEYS, function=b'\x10', params=keyindex, features_array=features_array)
|
||||
if keydata:
|
||||
key, key_task, flags = struct.unpack('!HHB', keydata[:5])
|
||||
keys.append(ReprogrammableKeyInfo(index, key, KEY_NAME[key], key_task, KEY_NAME[key_task], flags))
|
||||
key, key_task, flags = _unpack('!HHB', keydata[:5])
|
||||
keys.append(ReprogrammableKeyInfo(index, key, C.KEY_NAME[key], key_task, C.KEY_NAME[key_task], flags))
|
||||
|
||||
return keys
|
||||
|
|
|
|||
|
|
@ -4,11 +4,11 @@
|
|||
#
|
||||
|
||||
import logging
|
||||
import struct
|
||||
from binascii import hexlify
|
||||
from struct import pack as _pack
|
||||
from binascii import hexlify as _hexlify
|
||||
|
||||
from .constants import *
|
||||
from .exceptions import *
|
||||
import constants as C
|
||||
import exceptions as E
|
||||
|
||||
from . import unhandled as _unhandled
|
||||
import hidapi as _hid
|
||||
|
|
@ -92,9 +92,9 @@ def try_open(path):
|
|||
# any other replies are ignored, and will assume this is the wrong receiver device
|
||||
if reply == b'\x01\x00\x00\x00\x00\x00\x00\x00':
|
||||
# no idea what this is, but it comes up occasionally
|
||||
_l.log(_LOG_LEVEL, "[%s] (%d,) mistery reply [%s]", path, receiver_handle, hexlify(reply))
|
||||
_l.log(_LOG_LEVEL, "[%s] (%d,) mistery reply [%s]", path, receiver_handle, _hexlify(reply))
|
||||
else:
|
||||
_l.log(_LOG_LEVEL, "[%s] (%d,) unknown reply [%s]", path, receiver_handle, hexlify(reply))
|
||||
_l.log(_LOG_LEVEL, "[%s] (%d,) unknown reply [%s]", path, receiver_handle, _hexlify(reply))
|
||||
else:
|
||||
_l.log(_LOG_LEVEL, "[%s] (%d,) no reply", path, receiver_handle)
|
||||
|
||||
|
|
@ -146,16 +146,18 @@ def write(handle, device, data):
|
|||
data += b'\x00' * (_MIN_CALL_SIZE - 2 - len(data))
|
||||
elif len(data) > _MIN_CALL_SIZE - 2:
|
||||
data += b'\x00' * (_MAX_CALL_SIZE - 2 - len(data))
|
||||
wdata = struct.pack('!BB', 0x10, device) + data
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) <= w[%s]", handle, device, hexlify(wdata))
|
||||
wdata = _pack('!BB', 0x10, device) + data
|
||||
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) <= w[%s]", handle, device, _hexlify(wdata))
|
||||
if len(wdata) < _MIN_CALL_SIZE:
|
||||
_l.warn("(%d:%d) <= w[%s] call packet too short: %d bytes", handle, device, hexlify(wdata), len(wdata))
|
||||
_l.warn("(%d:%d) <= w[%s] call packet too short: %d bytes", handle, device, _hexlify(wdata), len(wdata))
|
||||
if len(wdata) > _MAX_CALL_SIZE:
|
||||
_l.warn("(%d:%d) <= w[%s] call packet too long: %d bytes", handle, device, hexlify(wdata), len(wdata))
|
||||
_l.warn("(%d:%d) <= w[%s] call packet too long: %d bytes", handle, device, _hexlify(wdata), len(wdata))
|
||||
|
||||
if not _hid.write(handle, wdata):
|
||||
_l.warn("(%d,%d) write failed, assuming receiver no longer available", handle, device)
|
||||
close(handle)
|
||||
raise NoReceiver()
|
||||
raise E.NoReceiver
|
||||
|
||||
|
||||
def read(handle, timeout=DEFAULT_TIMEOUT):
|
||||
|
|
@ -178,17 +180,18 @@ def read(handle, timeout=DEFAULT_TIMEOUT):
|
|||
if data is None:
|
||||
_l.warn("(%d,*) read failed, assuming receiver no longer available", handle)
|
||||
close(handle)
|
||||
raise NoReceiver
|
||||
raise E.NoReceiver
|
||||
|
||||
if data:
|
||||
_l.log(_LOG_LEVEL, "(%d,*) => r[%s]", handle, hexlify(data))
|
||||
_l.log(_LOG_LEVEL, "(%d,*) => r[%s]", handle, _hexlify(data))
|
||||
if len(data) < _MIN_REPLY_SIZE:
|
||||
_l.warn("(%d,*) => r[%s] read packet too short: %d bytes", handle, hexlify(data), len(data))
|
||||
_l.warn("(%d,*) => r[%s] read packet too short: %d bytes", handle, _hexlify(data), len(data))
|
||||
if len(data) > _MAX_REPLY_SIZE:
|
||||
_l.warn("(%d,*) => r[%s] read packet too long: %d bytes", handle, hexlify(data), len(data))
|
||||
_l.warn("(%d,*) => r[%s] read packet too long: %d bytes", handle, _hexlify(data), len(data))
|
||||
code = ord(data[:1])
|
||||
device = ord(data[1:2])
|
||||
return code, device, data[2:]
|
||||
|
||||
_l.log(_LOG_LEVEL, "(%d,*) => r[]", handle)
|
||||
|
||||
|
||||
|
|
@ -209,13 +212,16 @@ def request(handle, device, feature_index_function, params=b'', features_array=N
|
|||
available.
|
||||
:raisees FeatureCallError: if the feature call replied with an error.
|
||||
"""
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request {%s} params [%s]", handle, device, hexlify(feature_index_function), hexlify(params))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request {%s} params [%s]", handle, device, _hexlify(feature_index_function), _hexlify(params))
|
||||
if len(feature_index_function) != 2:
|
||||
raise ValueError('invalid feature_index_function {%s}: it must be a two-byte string' % hexlify(feature_index_function))
|
||||
raise ValueError('invalid feature_index_function {%s}: it must be a two-byte string' % _hexlify(feature_index_function))
|
||||
|
||||
retries = 5
|
||||
|
||||
write(handle, device, feature_index_function + params)
|
||||
while True:
|
||||
while retries > 0:
|
||||
reply = read(handle)
|
||||
retries -= 1
|
||||
|
||||
if not reply:
|
||||
# keep waiting...
|
||||
|
|
@ -225,7 +231,7 @@ def request(handle, device, feature_index_function, params=b'', features_array=N
|
|||
|
||||
if reply_device != device:
|
||||
# this message not for the device we're interested in
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request got reply for unexpected device %d: [%s]", handle, device, reply_device, hexlify(reply_data))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request got reply for unexpected device %d: [%s]", handle, device, reply_device, _hexlify(reply_data))
|
||||
# worst case scenario, this is a reply for a concurrent request
|
||||
# on this receiver
|
||||
_unhandled._publish(reply_code, reply_device, reply_data)
|
||||
|
|
@ -233,27 +239,27 @@ def request(handle, device, feature_index_function, params=b'', features_array=N
|
|||
|
||||
if reply_code == 0x10 and reply_data[:1] == b'\x8F' and reply_data[1:2] == feature_index_function:
|
||||
# device not present
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request ping failed on {%s} call: [%s]", handle, device, hexlify(feature_index_function), hexlify(reply_data))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request ping failed on {%s} call: [%s]", handle, device, _hexlify(feature_index_function), _hexlify(reply_data))
|
||||
return None
|
||||
|
||||
if reply_code == 0x10 and reply_data[:1] == b'\x8F':
|
||||
# device not present
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request ping failed: [%s]", handle, device, hexlify(reply_data))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) request ping failed: [%s]", handle, device, _hexlify(reply_data))
|
||||
return None
|
||||
|
||||
if reply_code == 0x11 and reply_data[0] == b'\xFF' and reply_data[1:3] == feature_index_function:
|
||||
# an error returned from the device
|
||||
error_code = ord(reply_data[3])
|
||||
_l.warn("(%d,%d) request feature call error %d = %s: %s", handle, device, error_code, ERROR_NAME[error_code], hexlify(reply_data))
|
||||
_l.warn("(%d,%d) request feature call error %d = %s: %s", handle, device, error_code, C.ERROR_NAME[error_code], _hexlify(reply_data))
|
||||
feature_index = ord(feature_index_function[:1])
|
||||
feature_function = feature_index_function[1:2]
|
||||
feature = None if features_array is None else features_array[feature_index]
|
||||
raise FeatureCallError(device, feature, feature_index, feature_function, error_code, reply_data)
|
||||
raise E.FeatureCallError(device, feature, feature_index, feature_function, error_code, reply_data)
|
||||
|
||||
if reply_code == 0x11 and reply_data[:2] == feature_index_function:
|
||||
# a matching reply
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) matched reply with data [%s]", handle, device, hexlify(reply_data[2:]))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) matched reply with feature-index-function [%s]", handle, device, _hexlify(reply_data[2:]))
|
||||
return reply_data[2:]
|
||||
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) unmatched reply {%s} (expected {%s})", handle, device, hexlify(reply_data[:2]), hexlify(feature_index_function))
|
||||
_l.log(_LOG_LEVEL, "(%d,%d) unmatched reply {%s} (expected {%s})", handle, device, _hexlify(reply_data[:2]), _hexlify(feature_index_function))
|
||||
_unhandled._publish(reply_code, reply_device, reply_data)
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@
|
|||
#
|
||||
|
||||
class FallbackDict(dict):
|
||||
def __init__(self, fallback_function, *args, **kwargs):
|
||||
def __init__(self, fallback_function=lambda x: None, *args, **kwargs):
|
||||
super(FallbackDict, self).__init__(*args, **kwargs)
|
||||
self.fallback = fallback_function
|
||||
|
||||
|
|
@ -18,7 +18,6 @@ def list2dict(values_list):
|
|||
return dict(zip(range(0, len(values_list)), values_list))
|
||||
|
||||
|
||||
|
||||
from collections import namedtuple
|
||||
|
||||
"""Tuple returned by list_devices and find_device_by_name."""
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@
|
|||
from binascii import hexlify as _hexlify
|
||||
from struct import pack as _pack
|
||||
|
||||
from .common import *
|
||||
from .common import (FallbackDict, list2dict)
|
||||
|
||||
|
||||
"""Possible features available on a Logitech device.
|
||||
|
|
@ -21,7 +21,7 @@ FEATURE = type('FEATURE', (),
|
|||
NAME=b'\x00\x05',
|
||||
BATTERY=b'\x10\x00',
|
||||
REPROGRAMMABLE_KEYS=b'\x1B\x00',
|
||||
WIRELESS_STATUS=b'\x1D\x4B',
|
||||
WIRELESS=b'\x1D\x4B',
|
||||
SOLAR_CHARGE=b'\x43\x01',
|
||||
))
|
||||
|
||||
|
|
@ -41,7 +41,7 @@ FEATURE_NAME[FEATURE.FIRMWARE] = 'FIRMWARE'
|
|||
FEATURE_NAME[FEATURE.NAME] = 'NAME'
|
||||
FEATURE_NAME[FEATURE.BATTERY] = 'BATTERY'
|
||||
FEATURE_NAME[FEATURE.REPROGRAMMABLE_KEYS] = 'REPROGRAMMABLE_KEYS'
|
||||
FEATURE_NAME[FEATURE.WIRELESS_STATUS] = 'WIRELESS_STATUS'
|
||||
FEATURE_NAME[FEATURE.WIRELESS] = 'WIRELESS'
|
||||
FEATURE_NAME[FEATURE.SOLAR_CHARGE] = 'SOLAR_CHARGE'
|
||||
|
||||
|
||||
|
|
@ -54,7 +54,7 @@ DEVICE_TYPE = FallbackDict(lambda x: 'unknown', list2dict(_DEVICE_TYPES))
|
|||
|
||||
_FIRMWARE_TYPES = ('Main (HID)', 'Bootloader', 'Hardware', 'Other')
|
||||
|
||||
"""Names of different firmware levels possible, ordered from top to bottom."""
|
||||
"""Names of different firmware levels possible, indexed by level."""
|
||||
FIRMWARE_TYPE = FallbackDict(lambda x: 'Unknown', list2dict(_FIRMWARE_TYPES))
|
||||
|
||||
|
||||
|
|
@ -74,7 +74,7 @@ _KEY_NAMES = ( 'unknown_0000', 'Volume up', 'Volume down', 'Mute', 'Play/Pause',
|
|||
KEY_NAME = FallbackDict(lambda x: 'unknown_%04x' % x, list2dict(_KEY_NAMES))
|
||||
|
||||
"""Possible flags on a reprogrammable key."""
|
||||
KEY_FLAG = type('REPROGRAMMABLE_KEY_FLAGS', (), dict(
|
||||
KEY_FLAG = type('KEY_FLAG', (), dict(
|
||||
REPROGRAMMABLE=0x10,
|
||||
FN_SENSITIVE=0x08,
|
||||
NONSTANDARD=0x04,
|
||||
|
|
@ -82,9 +82,20 @@ KEY_FLAG = type('REPROGRAMMABLE_KEY_FLAGS', (), dict(
|
|||
MSE=0x01,
|
||||
))
|
||||
|
||||
KEY_FLAG_NAME = FallbackDict(lambda x: 'unknown')
|
||||
KEY_FLAG_NAME[KEY_FLAG.REPROGRAMMABLE] = 'reprogrammable'
|
||||
KEY_FLAG_NAME[KEY_FLAG.FN_SENSITIVE] = 'fn-sensitive'
|
||||
KEY_FLAG_NAME[KEY_FLAG.NONSTANDARD] = 'nonstandard'
|
||||
KEY_FLAG_NAME[KEY_FLAG.IS_FN] = 'is-fn'
|
||||
KEY_FLAG_NAME[KEY_FLAG.MSE] = 'mse'
|
||||
|
||||
_ERROR_NAMES = ('Ok', 'Unknown', 'Invalid argument', 'Out of range',
|
||||
'Hardware error', 'Logitech internal', 'Invalid feature index',
|
||||
'Invalid function', 'Busy', 'Unsupported')
|
||||
|
||||
"""Names for error codes."""
|
||||
ERROR_NAME = FallbackDict(lambda x: 'Unknown error', list2dict(_ERROR_NAMES))
|
||||
|
||||
|
||||
del FallbackDict
|
||||
del list2dict
|
||||
|
|
|
|||
|
|
@ -2,8 +2,7 @@
|
|||
# Exceptions that may be raised by this API.
|
||||
#
|
||||
|
||||
from .constants import FEATURE_NAME as _FEATURE_NAME
|
||||
from .constants import ERROR_NAME as _ERROR_NAME
|
||||
from . import constants as C
|
||||
|
||||
|
||||
class NoReceiver(Exception):
|
||||
|
|
@ -17,21 +16,21 @@ class NoReceiver(Exception):
|
|||
class FeatureNotSupported(Exception):
|
||||
"""Raised when trying to request a feature not supported by the device."""
|
||||
def __init__(self, device, feature):
|
||||
super(FeatureNotSupported, self).__init__(device, feature, _FEATURE_NAME[feature])
|
||||
super(FeatureNotSupported, self).__init__(device, feature, C.FEATURE_NAME[feature])
|
||||
self.device = device
|
||||
self.feature = feature
|
||||
self.feature_name = _FEATURE_NAME[feature]
|
||||
self.feature_name = C.FEATURE_NAME[feature]
|
||||
|
||||
|
||||
class FeatureCallError(Exception):
|
||||
"""Raised if the device replied to a feature call with an error."""
|
||||
def __init__(self, device, feature, feature_index, feature_function, error_code, data=None):
|
||||
super(FeatureCallError, self).__init__(device, feature, feature_index, feature_function, error_code, _ERROR_NAME[error_code])
|
||||
super(FeatureCallError, self).__init__(device, feature, feature_index, feature_function, error_code, C.ERROR_NAME[error_code])
|
||||
self.device = device
|
||||
self.feature = feature
|
||||
self.feature_name = None if feature is None else _FEATURE_NAME[feature]
|
||||
self.feature_name = None if feature is None else C.FEATURE_NAME[feature]
|
||||
self.feature_index = feature_index
|
||||
self.feature_function = feature_function
|
||||
self.error_code = error_code
|
||||
self.error_string = _ERROR_NAME[error_code]
|
||||
self.error_string = C.ERROR_NAME[error_code]
|
||||
self.data = data
|
||||
|
|
|
|||
|
|
@ -4,11 +4,12 @@
|
|||
|
||||
import logging
|
||||
import threading
|
||||
from time import sleep
|
||||
from binascii import hexlify
|
||||
from time import sleep as _sleep
|
||||
# from binascii import hexlify as _hexlify
|
||||
|
||||
from . import base
|
||||
from .exceptions import *
|
||||
from . import base as _base
|
||||
from . import exceptions as E
|
||||
# from . import unhandled as _unhandled
|
||||
|
||||
|
||||
_LOG_LEVEL = 6
|
||||
|
|
@ -16,7 +17,7 @@ _l = logging.getLogger('logitech.unifying_receiver.listener')
|
|||
|
||||
|
||||
_READ_EVENT_TIMEOUT = 90 # ms
|
||||
_IDLE_SLEEP = 900 # ms
|
||||
_IDLE_SLEEP = 950 # ms
|
||||
|
||||
|
||||
class EventsListener(threading.Thread):
|
||||
|
|
@ -34,6 +35,7 @@ class EventsListener(threading.Thread):
|
|||
def __init__(self, receiver, events_callback):
|
||||
super(EventsListener, self).__init__(name='Unifying_Receiver_Listener_' + hex(receiver))
|
||||
self.daemon = True
|
||||
self.active = False
|
||||
|
||||
self.receiver = receiver
|
||||
self.callback = events_callback
|
||||
|
|
@ -45,13 +47,17 @@ class EventsListener(threading.Thread):
|
|||
self.task_done = threading.Event()
|
||||
|
||||
def run(self):
|
||||
_l.log(_LOG_LEVEL, "(%d) starting", self.receiver)
|
||||
self.active = True
|
||||
_l.log(_LOG_LEVEL, "(%d) starting", self.receiver)
|
||||
|
||||
# last_hook = _unhandled.hook
|
||||
# _unhandled.hook = self.callback
|
||||
|
||||
while self.active:
|
||||
try:
|
||||
# _l.log(_LOG_LEVEL, "(%d) reading next event", self.receiver)
|
||||
event = base.read(self.receiver, _READ_EVENT_TIMEOUT)
|
||||
except NoReceiver:
|
||||
event = _base.read(self.receiver, _READ_EVENT_TIMEOUT)
|
||||
except E.NoReceiver:
|
||||
_l.warn("(%d) receiver disconnected", self.receiver)
|
||||
self.active = False
|
||||
break
|
||||
|
|
@ -62,11 +68,13 @@ class EventsListener(threading.Thread):
|
|||
self.callback.__call__(*event)
|
||||
elif self.task is None:
|
||||
# _l.log(_LOG_LEVEL, "(%d) idle sleep", self.receiver)
|
||||
sleep(_IDLE_SLEEP / 1000.0)
|
||||
_sleep(_IDLE_SLEEP / 1000.0)
|
||||
else:
|
||||
self.task_reply = self._make_request(*self.task)
|
||||
self.task_done.set()
|
||||
|
||||
# _unhandled.hook = last_hook
|
||||
|
||||
def stop(self):
|
||||
"""Tells the listener to stop as soon as possible."""
|
||||
_l.log(_LOG_LEVEL, "(%d) stopping", self.receiver)
|
||||
|
|
@ -88,7 +96,7 @@ class EventsListener(threading.Thread):
|
|||
self.task = self.task_reply = None
|
||||
self.task_processing.release()
|
||||
|
||||
# _l.log(_LOG_LEVEL, "(%d) request '%s.%s' => [%s]", self.receiver, api_function.__module__, api_function.__name__, hexlify(reply))
|
||||
# _l.log(_LOG_LEVEL, "(%d) request '%s.%s' => [%s]", self.receiver, api_function.__module__, api_function.__name__, _hexlify(reply))
|
||||
if isinstance(reply, Exception):
|
||||
raise reply
|
||||
return reply
|
||||
|
|
@ -97,7 +105,7 @@ class EventsListener(threading.Thread):
|
|||
_l.log(_LOG_LEVEL, "(%d) calling '%s.%s' with %s, %s", self.receiver, api_function.__module__, api_function.__name__, args, kwargs)
|
||||
try:
|
||||
return api_function.__call__(self.receiver, *args, **kwargs)
|
||||
except NoReceiver as nr:
|
||||
except E.NoReceiver as nr:
|
||||
self.task_reply = nr
|
||||
self.active = False
|
||||
except Exception as e:
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@
|
|||
import unittest
|
||||
import struct
|
||||
|
||||
from logitech.unifying_receiver.constants import *
|
||||
from ..constants import *
|
||||
|
||||
|
||||
class Test_UR_Constants(unittest.TestCase):
|
||||
|
|
|
|||
|
|
@ -5,10 +5,10 @@
|
|||
import unittest
|
||||
from binascii import hexlify
|
||||
|
||||
from logitech.unifying_receiver import base
|
||||
from logitech.unifying_receiver.exceptions import *
|
||||
from logitech.unifying_receiver.constants import *
|
||||
from logitech.unifying_receiver.unhandled import *
|
||||
from .. import base
|
||||
from ..exceptions import *
|
||||
from ..constants import *
|
||||
from .. import unhandled
|
||||
|
||||
|
||||
class Test_UR_Base(unittest.TestCase):
|
||||
|
|
@ -157,15 +157,15 @@ class Test_UR_Base(unittest.TestCase):
|
|||
global received_unhandled
|
||||
received_unhandled = (code, device, data)
|
||||
|
||||
# set_unhandled_hook(_unhandled)
|
||||
unhandled.hook = _unhandled
|
||||
base.write(self.handle, self.device, FEATURE.ROOT + FEATURE.FEATURE_SET)
|
||||
reply = base.request(self.handle, self.device, fs_index + b'\x00')
|
||||
self.assertIsNotNone(reply, "request returned None reply")
|
||||
self.assertNotEquals(reply[:1], b'\x00')
|
||||
# self.assertIsNotNone(received_unhandled, "extra message not received by unhandled hook")
|
||||
self.assertIsNotNone(received_unhandled, "extra message not received by unhandled hook")
|
||||
|
||||
received_unhandled = None
|
||||
# set_unhandled_hook()
|
||||
unhandled.hook = None
|
||||
base.write(self.handle, self.device, FEATURE.ROOT + FEATURE.FEATURE_SET)
|
||||
reply = base.request(self.handle, self.device, fs_index + b'\x00')
|
||||
self.assertIsNotNone(reply, "request returned None reply")
|
||||
|
|
|
|||
|
|
@ -5,9 +5,9 @@
|
|||
import unittest
|
||||
import warnings
|
||||
|
||||
from logitech.unifying_receiver import api
|
||||
from logitech.unifying_receiver.exceptions import *
|
||||
from logitech.unifying_receiver.constants import *
|
||||
from .. import api
|
||||
from ..constants import *
|
||||
from ..common import *
|
||||
|
||||
|
||||
class Test_UR_API(unittest.TestCase):
|
||||
|
|
@ -45,7 +45,7 @@ class Test_UR_API(unittest.TestCase):
|
|||
|
||||
devices = []
|
||||
|
||||
for device in range(1, 1 + api.base.MAX_ATTACHED_DEVICES):
|
||||
for device in range(1, 1 + api._base.MAX_ATTACHED_DEVICES):
|
||||
ok = api.ping(self.handle, device)
|
||||
self.assertIsNotNone(ok, "invalid ping reply")
|
||||
if ok:
|
||||
|
|
@ -98,7 +98,7 @@ class Test_UR_API(unittest.TestCase):
|
|||
self.assertIsNotNone(d_firmware, "failed to get device firmware")
|
||||
self.assertGreater(len(d_firmware), 0, "device reported no firmware")
|
||||
for fw in d_firmware:
|
||||
self.assertIsInstance(fw, api.FirmwareInfo)
|
||||
self.assertIsInstance(fw, FirmwareInfo)
|
||||
|
||||
def test_52_get_device_type(self):
|
||||
if self.handle is None:
|
||||
|
|
@ -134,7 +134,7 @@ class Test_UR_API(unittest.TestCase):
|
|||
|
||||
device_info = api.get_device_info(self.handle, self.device, features_array=self.features_array)
|
||||
self.assertIsNotNone(device_info, "failed to read full device info")
|
||||
self.assertIsInstance(device_info, api.AttachedDeviceInfo)
|
||||
self.assertIsInstance(device_info, AttachedDeviceInfo)
|
||||
Test_UR_API.device_info = device_info
|
||||
|
||||
def test_60_get_battery_level(self):
|
||||
|
|
@ -160,7 +160,7 @@ class Test_UR_API(unittest.TestCase):
|
|||
if all_devices:
|
||||
self.assertIsNotNone(self.device)
|
||||
for device_info in all_devices:
|
||||
self.assertIsInstance(device_info, api.AttachedDeviceInfo)
|
||||
self.assertIsInstance(device_info, AttachedDeviceInfo)
|
||||
else:
|
||||
self.assertIsNone(self.device)
|
||||
|
||||
|
|
|
|||
|
|
@ -4,27 +4,16 @@
|
|||
#
|
||||
|
||||
import logging
|
||||
from binascii import hexlify
|
||||
|
||||
_l = logging.getLogger('logitech.unifying_receiver.unhandled')
|
||||
from binascii import hexlify as _hexlify
|
||||
|
||||
|
||||
def _logging_unhandled_hook(reply_code, device, data):
|
||||
def _logdebug_hook(reply_code, device, data):
|
||||
"""Default unhandled hook, logs the reply as DEBUG."""
|
||||
_l.debug("UNHANDLED (,%d) code 0x%02x data [%s]", device, reply_code, hexlify(data))
|
||||
_l = logging.getLogger('logitech.unifying_receiver.unhandled')
|
||||
_l.debug("UNHANDLED (,%d) code 0x%02x data [%s]", device, reply_code, _hexlify(data))
|
||||
|
||||
|
||||
_unhandled_hook = _logging_unhandled_hook
|
||||
|
||||
|
||||
def _publish(reply_code, device, data):
|
||||
"""Delivers a reply to the unhandled hook, if any."""
|
||||
if _unhandled_hook is not None:
|
||||
_unhandled_hook.__call__(reply_code, device, data)
|
||||
|
||||
|
||||
def set_unhandled_hook(hook=None):
|
||||
"""Sets the function that will be called on unhandled incoming events.
|
||||
"""The function that will be called on unhandled incoming events.
|
||||
|
||||
The hook must be a function with the signature: ``_(int, int, str)``, where
|
||||
the parameters are: (reply code, device number, data).
|
||||
|
|
@ -38,5 +27,10 @@ def set_unhandled_hook(hook=None):
|
|||
|
||||
The default implementation logs the unhandled reply as DEBUG.
|
||||
"""
|
||||
global _unhandled_hook
|
||||
_unhandled_hook = hook
|
||||
hook = _logdebug_hook
|
||||
|
||||
|
||||
def _publish(reply_code, device, data):
|
||||
"""Delivers a reply to the unhandled hook, if any."""
|
||||
if hook is not None:
|
||||
hook.__call__(reply_code, device, data)
|
||||
|
|
|
|||
5
solaar
5
solaar
|
|
@ -2,7 +2,8 @@
|
|||
|
||||
cd `dirname "$0"`
|
||||
|
||||
export LD_LIBRARY_PATH=$PWD/lib
|
||||
export PYTHONPATH=$PWD/lib
|
||||
export LD_LIBRARY_PATH=$LD_LIBRARY_PATH:$PWD/lib
|
||||
export PYTHONPATH=$PWD:$PWD/lib
|
||||
|
||||
exec python -OO solaar.py "$@"
|
||||
#exec python -OO -m profile -o $TMPDIR/profile.log solaar.py "$@"
|
||||
|
|
|
|||
277
solaar.py
277
solaar.py
|
|
@ -1,268 +1,14 @@
|
|||
#!/usr/bin/env python
|
||||
|
||||
__version__ = '0.3'
|
||||
__version__ = '0.4'
|
||||
|
||||
#
|
||||
#
|
||||
#
|
||||
|
||||
import logging
|
||||
import time
|
||||
import threading
|
||||
|
||||
from gi.repository import GObject
|
||||
from gi.repository import Gtk
|
||||
|
||||
from logitech.unifying_receiver import api as ur
|
||||
from logitech.unifying_receiver.listener import EventsListener
|
||||
|
||||
from logitech.devices import *
|
||||
|
||||
#
|
||||
# A few constants
|
||||
#
|
||||
|
||||
APP_TITLE = 'Solaar'
|
||||
UNIFYING_RECEIVER = 'Unifying Receiver'
|
||||
NO_DEVICES = 'No devices attached.'
|
||||
NO_RECEIVER = 'Unifying Receiver not found.'
|
||||
FOUND_RECEIVER = 'Unifying Receiver found.'
|
||||
|
||||
STATUS_TIMEOUT = 61 # seconds
|
||||
ICON_UPDATE_SLEEP = 7 # seconds
|
||||
|
||||
#
|
||||
# Optional desktop notifications.
|
||||
#
|
||||
|
||||
try:
|
||||
import notify2
|
||||
notify2.init(APP_TITLE)
|
||||
|
||||
_notifications = {}
|
||||
_ICONS = {}
|
||||
import os.path
|
||||
|
||||
def notify_desktop(status, title, text, icon=None):
|
||||
logging.debug("notify_desktop [%d] %s: %s", status, title, text)
|
||||
def _icon_path(name):
|
||||
path = os.path.join(__file__, '..', 'images', name + '.png')
|
||||
path = os.path.abspath(os.path.normpath(path))
|
||||
return path if os.path.isfile(path) else None
|
||||
|
||||
if icon is None:
|
||||
icon = title
|
||||
if icon in _ICONS:
|
||||
path = _ICONS[icon]
|
||||
else:
|
||||
_ICONS[icon] = path = _icon_path(icon)
|
||||
if path:
|
||||
icon = path
|
||||
if icon is None:
|
||||
icon = 'error' if status < 0 else 'info'
|
||||
|
||||
if title in _notifications:
|
||||
notification = _notifications[title]
|
||||
else:
|
||||
_notifications[title] = notification = notify2.Notification(title, icon=icon)
|
||||
notification.set_category(APP_TITLE)
|
||||
|
||||
notification.set_urgency(notify2.URGENCY_CRITICAL if status < 0 else notify2.URGENCY_NORMAL)
|
||||
notification.update(title, text, icon)
|
||||
notification.show()
|
||||
|
||||
def clear_notifications():
|
||||
all(n.close() for n in list(_notifications.values()))
|
||||
notify2.uninit()
|
||||
_notifications.clear()
|
||||
_ICONS.clear()
|
||||
|
||||
except ImportError:
|
||||
logging.warn("python-notify2 not found, desktop notifications are disabled")
|
||||
def notify_desktop(status, title, text, icon=None): pass
|
||||
def clear_notifications(): pass
|
||||
|
||||
#
|
||||
#
|
||||
#
|
||||
|
||||
class StatusThread(threading.Thread):
|
||||
def __init__(self, status_icon):
|
||||
super(StatusThread, self).__init__(name='StatusThread')
|
||||
self.daemon = True
|
||||
self.status_icon = status_icon
|
||||
|
||||
self.last_receiver_status = None
|
||||
self.listener = None
|
||||
self.devices = {}
|
||||
self.statuses = {}
|
||||
|
||||
def run(self):
|
||||
self.active = True
|
||||
while self.active:
|
||||
if self.listener is None:
|
||||
receiver = ur.open()
|
||||
if receiver:
|
||||
for devinfo in ur.list_devices(receiver):
|
||||
self.devices[devinfo.number] = devinfo
|
||||
self.listener = EventsListener(receiver, self.events_callback)
|
||||
self.listener.start()
|
||||
logging.info(str(self.listener))
|
||||
notify_desktop(1, UNIFYING_RECEIVER, FOUND_RECEIVER)
|
||||
self.last_receiver_status = 1
|
||||
else:
|
||||
if self.last_receiver_status != -1:
|
||||
notify_desktop(-1, UNIFYING_RECEIVER, NO_RECEIVER)
|
||||
self.last_receiver_status = -1
|
||||
elif not self.listener.active:
|
||||
logging.info(str(self.listener))
|
||||
self.listener = None
|
||||
self.devices.clear()
|
||||
self.statuses.clear()
|
||||
notify_desktop(-1, UNIFYING_RECEIVER, NO_RECEIVER)
|
||||
self.last_receiver_status = -1
|
||||
|
||||
if self.active:
|
||||
update_icon = True
|
||||
if self.listener and self.devices:
|
||||
update_icon &= self.update_old_statuses()
|
||||
|
||||
if self.active:
|
||||
if update_icon:
|
||||
GObject.idle_add(self.update_status_icon)
|
||||
time.sleep(ICON_UPDATE_SLEEP)
|
||||
|
||||
def stop(self):
|
||||
self.active = False
|
||||
if self.listener:
|
||||
self.listener.stop()
|
||||
ur.close(self.listener.receiver)
|
||||
|
||||
def update_old_statuses(self):
|
||||
updated = False
|
||||
|
||||
for devinfo in list(self.devices.values()):
|
||||
if devinfo.number in self.statuses:
|
||||
last_status_time = self.statuses[devinfo.number][0]
|
||||
if time.time() - last_status_time > STATUS_TIMEOUT:
|
||||
status = request_status(devinfo, self.listener)
|
||||
updated |= self.device_status_changed(devinfo, status)
|
||||
else:
|
||||
self.statuses[devinfo.number] = [0, None, None]
|
||||
updated |= self.device_status_changed(devinfo, (DEVICE_STATUS.CONNECTED, None))
|
||||
|
||||
return updated
|
||||
|
||||
def events_callback(self, code, device, data):
|
||||
updated = False
|
||||
|
||||
if device in self.devices:
|
||||
devinfo = self.devices[device]
|
||||
if code == 0x10 and data[0] == 'b\x8F':
|
||||
updated = True
|
||||
self.device_status_changed(devinfo, DEVICE_STATUS.UNAVAILABLE)
|
||||
elif code == 0x11:
|
||||
status = process_event(devinfo, self.listener, data)
|
||||
updated |= self.device_status_changed(devinfo, status)
|
||||
else:
|
||||
logging.warn("unknown event code %02x", code)
|
||||
elif device:
|
||||
logging.debug("got event (%d, %d, %s) for new device", code, device, data)
|
||||
devinfo = ur.get_device_info(self.listener.receiver, device)
|
||||
if devinfo:
|
||||
self.devices[device] = devinfo
|
||||
self.statuses[device] = [0, None, None]
|
||||
updated |= self.device_status_changed(devinfo, (DEVICE_STATUS.CONNECTED, None))
|
||||
# else:
|
||||
# logging.warn("got event (%d, %d, %s) for unknown device", code, device, data)
|
||||
else:
|
||||
logging.warn("don't know how to handle event (%d, %d, %s)", code, device, data)
|
||||
|
||||
if updated:
|
||||
GObject.idle_add(self.update_status_icon)
|
||||
|
||||
def device_status_changed(self, devinfo, status):
|
||||
if status is None or devinfo.number not in self.statuses:
|
||||
return False
|
||||
|
||||
if type(status) == int:
|
||||
status_code = status
|
||||
status_text = DEVICE_STATUS_NAME[status_code]
|
||||
else:
|
||||
status_code = status[0]
|
||||
status_text = DEVICE_STATUS_NAME[status_code] if status[1] is None else status[1]
|
||||
|
||||
device_status = self.statuses[devinfo.number]
|
||||
old_status_code = device_status[1]
|
||||
|
||||
device_status[0] = time.time()
|
||||
device_status[1] = status_code
|
||||
device_status[2] = status_text
|
||||
|
||||
if old_status_code == status_code:
|
||||
# the device has been missing twice in a row (1 to 2 minutes), so
|
||||
# forget about it
|
||||
if status_code == DEVICE_STATUS.UNAVAILABLE:
|
||||
del self.devices[devinfo.number]
|
||||
del self.statuses[devinfo.number]
|
||||
else:
|
||||
logging.debug("device status changed from %s => %s: %s", old_status_code, status_code, status_text)
|
||||
notify_desktop(status_code, devinfo.name, status_text)
|
||||
|
||||
return True
|
||||
|
||||
def update_status_icon(self):
|
||||
if self.listener and self.listener.active:
|
||||
if self.devices:
|
||||
all_statuses = []
|
||||
for d in self.devices:
|
||||
devinfo = self.devices[d]
|
||||
status_text = self.statuses[d][2]
|
||||
if status_text:
|
||||
if ' ' in status_text:
|
||||
all_statuses.append(devinfo.name)
|
||||
all_statuses.append(' ' + status_text)
|
||||
else:
|
||||
all_statuses.append(devinfo.name + ' ' + status_text)
|
||||
else:
|
||||
all_statuses.append(devinfo.name)
|
||||
tooltip = '\n'.join(all_statuses)
|
||||
else:
|
||||
tooltip = NO_DEVICES
|
||||
else:
|
||||
tooltip = NO_RECEIVER
|
||||
|
||||
self.status_icon.set_tooltip_text(tooltip)
|
||||
|
||||
def activate_icon(self, _):
|
||||
if self.listener and self.listener.active:
|
||||
if self.devices:
|
||||
for devinfo in list(self.devices.values()):
|
||||
_, status_code, status_text = self.statuses[devinfo.number]
|
||||
notify_desktop(status_code, devinfo.name, status_text)
|
||||
else:
|
||||
notify_desktop(1, UNIFYING_RECEIVER, NO_DEVICES)
|
||||
self.last_receiver_status = 1
|
||||
else:
|
||||
notify_desktop(-1, UNIFYING_RECEIVER, NO_RECEIVER)
|
||||
self.last_receiver_status = -1
|
||||
|
||||
|
||||
|
||||
def show_icon_menu(icon, button, time, status_thread):
|
||||
menu = Gtk.Menu()
|
||||
|
||||
status_item = Gtk.MenuItem('Status')
|
||||
status_item.connect('activate', status_thread.activate_icon)
|
||||
menu.append(status_item)
|
||||
|
||||
quit_item = Gtk.MenuItem('Quit')
|
||||
quit_item.connect('activate', Gtk.main_quit)
|
||||
menu.append(quit_item)
|
||||
|
||||
menu.show_all()
|
||||
menu.popup(None, None, icon.position_menu, icon, button, time)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
import argparse
|
||||
|
|
@ -272,21 +18,10 @@ if __name__ == '__main__':
|
|||
args = arg_parser.parse_args()
|
||||
|
||||
log_level = logging.root.level - 10 * args.verbose
|
||||
logging.root.setLevel(log_level if log_level > 0 else 1)
|
||||
logging.basicConfig(level=log_level if log_level > 0 else 1)
|
||||
|
||||
status_icon = Gtk.StatusIcon.new_from_file('images/' + UNIFYING_RECEIVER + '.png')
|
||||
status_icon.set_title(APP_TITLE)
|
||||
status_icon.set_name(APP_TITLE)
|
||||
status_icon.set_tooltip_text('Searching...')
|
||||
notify_desktop(0, UNIFYING_RECEIVER, 'Searching...')
|
||||
images_path = os.path.join(__file__, '..', 'images')
|
||||
images_path = os.path.abspath(os.path.normpath(images_path))
|
||||
|
||||
GObject.threads_init()
|
||||
status_thread = StatusThread(status_icon)
|
||||
status_thread.start()
|
||||
|
||||
status_icon.connect('popup_menu', show_icon_menu, status_thread)
|
||||
status_icon.connect('activate', status_thread.activate_icon)
|
||||
Gtk.main()
|
||||
|
||||
status_thread.stop()
|
||||
clear_notifications()
|
||||
import app
|
||||
app.run(images_path)
|
||||
|
|
|
|||
Loading…
Reference in New Issue