# -*- python-mode -*- ## Copyright (C) 2012-2013 Daniel Pavel ## ## This program is free software; you can redistribute it and/or modify ## it under the terms of the GNU General Public License as published by ## the Free Software Foundation; either version 2 of the License, or ## (at your option) any later version. ## ## This program is distributed in the hope that it will be useful, ## but WITHOUT ANY WARRANTY; without even the implied warranty of ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the ## GNU General Public License for more details. ## ## You should have received a copy of the GNU General Public License along ## with this program; if not, write to the Free Software Foundation, Inc., ## 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. import errno as _errno from logging import INFO as _INFO from logging import getLogger import hidapi as _hid from . import base as _base from . import hidpp10 as _hidpp10 from .base_usb import product_information as _product_information from .common import strhex as _strhex from .device import Device _log = getLogger(__name__) del getLogger _R = _hidpp10.REGISTERS _IR = _hidpp10.INFO_SUBREGISTERS # # # class Receiver: """A Unifying Receiver instance. The paired devices are available through the sequence interface. """ number = 0xFF kind = None def __init__(self, handle, path, product_id): assert handle self.isDevice = False # some devices act as receiver so we need a property to distinguish them self.handle = handle self.path = path self.product_id = product_id product_info = _product_information(self.product_id) if not product_info: _log.warning('Unknown receiver type: %s', self.product_id) product_info = {} self.receiver_kind = product_info.get('receiver_kind', 'unknown') # read the serial immediately, so we can find out max_devices self.last_id = 0 if self.receiver_kind == 'bolt': serial_reply = self.read_register(_R.bolt_uniqueId) self.serial = _strhex(serial_reply) self.max_devices = product_info.get('max_devices', 1) self.may_unpair = product_info.get('may_unpair', False) else: serial_reply = self.read_register(_R.receiver_info, _IR.receiver_information) if serial_reply: self.serial = _strhex(serial_reply[1:5]) self.max_devices = ord(serial_reply[6:7]) self.last_id = ord(serial_reply[7:8]) if self.max_devices <= 0 or self.max_devices > 6: self.max_devices = product_info.get('max_devices', 1) self.may_unpair = product_info.get('may_unpair', False) else: # handle receivers that don't have a serial number specially (i.e., c534 and Bolt receivers) self.serial = None self.max_devices = product_info.get('max_devices', 1) self.may_unpair = product_info.get('may_unpair', False) self.last_id = self.last_id if self.last_id else self.max_devices self.name = product_info.get('name', 'Receiver') self.re_pairs = product_info.get('re_pairs', False) self._str = '<%s(%s,%s%s)>' % ( self.name.replace(' ', ''), self.path, '' if isinstance(self.handle, int) else 'T', self.handle ) self._firmware = None self._devices = {} self._remaining_pairings = None def close(self): handle, self.handle = self.handle, None for _n, d in self._devices.items(): if d: d.close() self._devices.clear() return (handle and _base.close(handle)) def __del__(self): self.close() @property def firmware(self): if self._firmware is None and self.handle: self._firmware = _hidpp10.get_firmware(self) return self._firmware # how many pairings remain (None for unknown, -1 for unlimited) def remaining_pairings(self, cache=True): if self._remaining_pairings is None or not cache: ps = self.read_register(_R.receiver_connection) if ps is not None: ps = ord(ps[2:3]) self._remaining_pairings = ps - 5 if ps >= 5 else -1 return self._remaining_pairings def enable_connection_notifications(self, enable=True): """Enable or disable device (dis)connection notifications on this receiver.""" if not self.handle: return False if enable: set_flag_bits = ( _hidpp10.NOTIFICATION_FLAG.battery_status | _hidpp10.NOTIFICATION_FLAG.wireless | _hidpp10.NOTIFICATION_FLAG.software_present ) else: set_flag_bits = 0 ok = _hidpp10.set_notification_flags(self, set_flag_bits) if ok is None: _log.warn('%s: failed to %s receiver notifications', self, 'enable' if enable else 'disable') return None flag_bits = _hidpp10.get_notification_flags(self) flag_names = None if flag_bits is None else tuple(_hidpp10.NOTIFICATION_FLAG.flag_names(flag_bits)) if _log.isEnabledFor(_INFO): _log.info('%s: receiver notifications %s => %s', self, 'enabled' if enable else 'disabled', flag_names) return flag_bits def device_codename(self, n): if self.receiver_kind == 'bolt': codename = self.read_register(_R.receiver_info, _IR.bolt_device_name + n, 0x01) if codename: codename = codename[3:3 + min(14, ord(codename[2:3]))] return codename.decode('ascii') return codename = self.read_register(_R.receiver_info, _IR.device_name + n - 1) if codename: codename = codename[2:2 + ord(codename[1:2])] return codename.decode('ascii') def device_pairing_information(self, n): if self.receiver_kind == 'bolt': pair_info = self.read_register(_R.receiver_info, _IR.bolt_pairing_information + n) if pair_info: wpid = _strhex(pair_info[3:4]) + _strhex(pair_info[2:3]) kind = _hidpp10.DEVICE_KIND[ord(pair_info[1:2]) & 0x0F] return wpid, kind, 0 else: raise _base.NoSuchDevice(number=n, receiver=self, error='read Bolt wpid') wpid = 0 kind = None polling_rate = None pair_info = self.read_register(_R.receiver_info, _IR.pairing_information + n - 1) if pair_info: # may be either a Unifying receiver, or an Unifying-ready receiver wpid = _strhex(pair_info[3:5]) kind = _hidpp10.DEVICE_KIND[ord(pair_info[7:8]) & 0x0F] polling_rate = ord(pair_info[2:3]) elif self.receiver_kind == '27Mz': # 27Mhz receiver, fill extracting WPID from udev path wpid = _hid.find_paired_node_wpid(self.path, n) if not wpid: _log.error('Unable to get wpid from udev for device %d of %s', n, self) raise _base.NoSuchDevice(number=n, receiver=self, error='Not present 27Mhz device') kind = _hidpp10.DEVICE_KIND[self.get_kind_from_index(n, self)] else: # unifying protocol not supported, may be an old Nano receiver device_info = self.read_register(_R.receiver_info, 0x04) if device_info: wpid = _strhex(device_info[3:5]) kind = _hidpp10.DEVICE_KIND[0x00] # unknown kind return wpid, kind, polling_rate def device_extended_pairing_information(self, n): serial = None power_switch = '(unknown)' if self.receiver_kind == 'bolt': pair_info = self.read_register(_R.receiver_info, _IR.bolt_pairing_information + n) if pair_info: serial = _strhex(pair_info[4:8]) return serial, power_switch else: return '?', power_switch pair_info = self.read_register(_R.receiver_info, _IR.extended_pairing_information + n - 1) if pair_info: power_switch = _hidpp10.POWER_SWITCH_LOCATION[ord(pair_info[9:10]) & 0x0F] else: # some Nano receivers? pair_info = self.read_register(0x2D5) if pair_info: serial = _strhex(pair_info[1:5]) return serial, power_switch def get_kind_from_index(self, index): """Get device kind from 27Mhz device index""" # accordingly to drivers/hid/hid-logitech-dj.c # index 1 or 2 always mouse, index 3 always the keyboard, # index 4 is used for an optional separate numpad if index == 1: # mouse kind = 2 elif index == 2: # mouse kind = 2 elif index == 3: # keyboard kind = 1 elif index == 4: # numpad kind = 3 else: # unknown device number on 27Mhz receiver _log.error('failed to calculate device kind for device %d of %s', index, self) raise _base.NoSuchDevice(number=index, receiver=self, error='Unknown 27Mhz device number') return kind def notify_devices(self): """Scan all devices.""" if self.handle: if not self.write_register(_R.receiver_connection, 0x02): _log.warn('%s: failed to trigger device link notifications', self) def register_new_device(self, number, notification=None): if self._devices.get(number) is not None: raise IndexError('%s: device number %d already registered' % (self, number)) assert notification is None or notification.devnumber == number assert notification is None or notification.sub_id == 0x41 try: dev = Device(self, number, notification) if _log.isEnabledFor(_INFO): _log.info('%s: found new device %d (%s)', self, number, dev.wpid) self._devices[number] = dev return dev except _base.NoSuchDevice: _log.exception('register_new_device') _log.warning('%s: looked for device %d, not found', self, number) self._devices[number] = None def set_lock(self, lock_closed=True, device=0, timeout=0): if self.handle: action = 0x02 if lock_closed else 0x01 reply = self.write_register(_R.receiver_pairing, action, device, timeout) if reply: return True _log.warn('%s: failed to %s the receiver lock', self, 'close' if lock_closed else 'open') def discover(self, cancel=False, timeout=30): # Bolt device discovery assert self.receiver_kind == 'bolt' if self.handle: action = 0x02 if cancel else 0x01 reply = self.write_register(_R.bolt_device_discovery, timeout, action) if reply: return True _log.warn('%s: failed to %s device discovery', self, 'cancel' if cancel else 'start') def pair_device(self, pair=True, slot=0, address=b'\0\0\0\0\0\0', authentication=0x00, entropy=20): # Bolt pairing assert self.receiver_kind == 'bolt' if self.handle: action = 0x01 if pair is True else 0x03 if pair is False else 0x02 reply = self.write_register(_R.bolt_pairing, action, slot, address, authentication, entropy) if reply: return True _log.warn('%s: failed to %s device %s', self, 'pair' if pair else 'unpair', address) def count(self): count = self.read_register(_R.receiver_connection) return 0 if count is None else ord(count[1:2]) # def has_devices(self): # return len(self) > 0 or self.count() > 0 def request(self, request_id, *params): if bool(self): return _base.request(self.handle, 0xFF, request_id, *params) read_register = _hidpp10.read_register write_register = _hidpp10.write_register def __iter__(self): for number in range(1, 16): # some receivers have devices past their max # devices if number in self._devices: dev = self._devices[number] else: dev = self.__getitem__(number) if dev is not None: yield dev def __getitem__(self, key): if not bool(self): return None dev = self._devices.get(key) if dev is not None: return dev if not isinstance(key, int): raise TypeError('key must be an integer') if key < 1 or key > 15: # some receivers have devices past their max # devices raise IndexError(key) return self.register_new_device(key) def __delitem__(self, key): self._unpair_device(key, False) def _unpair_device(self, key, force=False): key = int(key) if self._devices.get(key) is None: raise IndexError(key) dev = self._devices[key] if not dev: if key in self._devices: del self._devices[key] return if self.re_pairs and not force: # invalidate the device, but these receivers don't unpair per se dev.online = False dev.wpid = None if key in self._devices: del self._devices[key] _log.warn('%s removed device %s', self, dev) else: if self.receiver_kind == 'bolt': reply = self.write_register(_R.bolt_pairing, 0x03, key) else: reply = self.write_register(_R.receiver_pairing, 0x03, key) if reply: # invalidate the device dev.online = False dev.wpid = None if key in self._devices: del self._devices[key] if _log.isEnabledFor(_INFO): _log.info('%s unpaired device %s', self, dev) else: _log.error('%s failed to unpair device %s', self, dev) raise Exception('failed to unpair device %s: %s' % (dev.name, key)) def __len__(self): return len([d for d in self._devices.values() if d is not None]) def __contains__(self, dev): if isinstance(dev, int): return self._devices.get(dev) is not None return self.__contains__(dev.number) def __eq__(self, other): return other is not None and self.kind == other.kind and self.path == other.path def __ne__(self, other): return other is None or self.kind != other.kind or self.path != other.path def __hash__(self): return self.path.__hash__() def __str__(self): return self._str __repr__ = __str__ __bool__ = __nonzero__ = lambda self: self.handle is not None @classmethod def open(self, device_info): """Opens a Logitech Receiver found attached to the machine, by Linux device path. :returns: An open file handle for the found receiver, or ``None``. """ try: handle = _base.open_path(device_info.path) if handle: return Receiver(handle, device_info.path, device_info.product_id) except OSError as e: _log.exception('open %s', device_info) if e.errno == _errno.EACCES: raise except Exception: _log.exception('open %s', device_info)