diff --git a/lib/logitech_receiver/centurion.py b/lib/logitech_receiver/centurion.py new file mode 100644 index 00000000..81d29e42 --- /dev/null +++ b/lib/logitech_receiver/centurion.py @@ -0,0 +1,510 @@ +## Copyright (C) 2012-2013 Daniel Pavel +## Copyright (C) 2014-2024 Solaar Contributors https://pwr-solaar.github.io/Solaar/ +## +## 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. + +"""Centurion device protocol — receiver class, factory, device info/firmware/battery queries. + +CenturionReceiver is a lightweight receiver-like container for Centurion +(PRO X 2 LIGHTSPEED and similar) dongles. Protocol functions query device +info, firmware, serial, name, and battery via Centurion-specific HID++ features. +""" + +from __future__ import annotations + +import errno +import logging +import struct + +from typing import Callable + +from solaar import configuration + +from . import base +from . import exceptions +from . import hidpp10 +from .common import Alert +from .common import Battery +from .common import BatteryStatus +from .common import FirmwareKind +from .common import _read_usb_product_string +from .hidpp20_constants import SupportedFeature + +logger = logging.getLogger(__name__) + + +# --- Centurion protocol functions (standalone, operate on any device-like object) --- + + +def get_firmware_centurion(device): + """Reads firmware info from a Centurion device via DeviceInfo (0x0100) function 1.""" + from . import common + + fw = [] + seen = set() # track response signatures to detect duplicates + for index in range(0, 8): # try up to 8 entities + try: + report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO, 0x10, index) + except exceptions.FeatureCallError: + break + if not report or len(report) < 5: + break + # Dedup: parent device returns the same response for every entity index + sig = bytes(report[: 5 + report[4]]) + if sig in seen: + break + seen.add(sig) + fw_type = report[0] + version = struct.unpack("!H", report[2:4])[0] + name_len = report[4] + name = report[5 : 5 + name_len].decode("ascii", errors="replace").rstrip("\x00") if name_len else "" + version_str = f"{version >> 8}.{version & 0xFF:02d}" + kind = FirmwareKind(fw_type) if fw_type <= 3 else FirmwareKind.Other + fw.append(common.FirmwareInfo(kind, name, version_str, None)) + return tuple(fw) if fw else None + + +def get_serial_centurion(device): + """Reads the serial number from a Centurion device via DeviceInfo (0x0100) function 2.""" + try: + report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO, 0x20) + except exceptions.FeatureCallError: + return None + if not report or len(report) < 2: + return None + str_len = report[0] + return report[1 : 1 + str_len].decode("ascii", errors="replace").rstrip("\x00") + + +def get_hardware_info_centurion(device): + """Reads hardware info from a Centurion device via DeviceInfo (0x0100) function 0. + + Returns (modelId, hardwareRevision, productId) or None. + """ + try: + report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO) + except exceptions.FeatureCallError: + return None + if not report or len(report) < 4: + return None + model_id = report[0] + hw_revision = report[1] + product_id = struct.unpack("!H", report[2:4])[0] + return model_id, hw_revision, product_id + + +def _centurion_sub_device_info_request(device, function=0x00, *params): + """Send a DeviceInfo (0x0100) request to the sub-device via bridge.""" + sub_indices = getattr(device, "_centurion_sub_indices", {}) + sub_idx = sub_indices.get(SupportedFeature.CENTURION_DEVICE_INFO) + if sub_idx is None: + return None + return device.centurion_bridge_request(sub_idx, function, *params) + + +def get_firmware_centurion_sub(device): + """Reads firmware info from the Centurion sub-device (headset) via bridge.""" + from . import common + + fw = [] + seen = set() + for index in range(0, 8): + report = _centurion_sub_device_info_request(device, 0x10, index) + if not report or len(report) < 5: + break + sig = bytes(report[: 5 + report[4]]) + if sig in seen: + break + seen.add(sig) + fw_type = report[0] + version = struct.unpack("!H", report[2:4])[0] + name_len = report[4] + name = report[5 : 5 + name_len].decode("ascii", errors="replace").rstrip("\x00") if name_len else "" + version_str = f"{version >> 8}.{version & 0xFF:02d}" + kind = FirmwareKind(fw_type) if fw_type <= 3 else FirmwareKind.Other + fw.append(common.FirmwareInfo(kind, name, version_str, None)) + return tuple(fw) if fw else None + + +def get_serial_centurion_sub(device): + """Reads the serial number from the Centurion sub-device (headset) via bridge.""" + report = _centurion_sub_device_info_request(device, 0x20) + if not report or len(report) < 2: + return None + str_len = report[0] + return report[1 : 1 + str_len].decode("ascii", errors="replace").rstrip("\x00") + + +def get_hardware_info_centurion_sub(device): + """Reads hardware info from the Centurion sub-device (headset) via bridge. + + Returns (modelId, hardwareRevision, productId) or None. + """ + report = _centurion_sub_device_info_request(device) + if not report or len(report) < 4: + return None + model_id = report[0] + hw_revision = report[1] + product_id = struct.unpack("!H", report[2:4])[0] + return model_id, hw_revision, product_id + + +def get_name_centurion(device): + """Reads a Centurion device's name via DeviceName (0x0101). + + Tries two response formats: + 1. Inline: function 0 returns [name_len, name_bytes...] (like serial) + 2. Chunked: function 0 returns [name_len], function 1 returns [name_bytes...] (like standard DeviceName) + """ + try: + reply = device.feature_request(SupportedFeature.CENTURION_DEVICE_NAME) + except exceptions.FeatureCallError: + return None + if not reply: + return None + name_length = reply[0] + if name_length == 0: + return None + # If the full name is inline (length + name bytes in one response) + if len(reply) >= 1 + name_length: + return reply[1 : 1 + name_length].decode("utf-8", errors="replace").rstrip("\x00") + # Otherwise, fetch name in chunks via function 1 (like standard DEVICE_NAME) + name = b"" + while len(name) < name_length: + try: + fragment = device.feature_request(SupportedFeature.CENTURION_DEVICE_NAME, 0x10, len(name)) + except exceptions.FeatureCallError: + break + if fragment: + name += fragment[: name_length - len(name)] + else: + break + return name.decode("utf-8", errors="replace").rstrip("\x00") if name else None + + +def get_battery_centurion(device): + """Query battery via CENTURION_BATTERY_SOC.""" + try: + report = device.feature_request(SupportedFeature.CENTURION_BATTERY_SOC) + if report is not None: + return decipher_battery_centurion(report) + except exceptions.FeatureCallError: + if SupportedFeature.CENTURION_BATTERY_SOC in device.features: + return SupportedFeature.CENTURION_BATTERY_SOC + return None + + +def decipher_battery_centurion(report) -> tuple[SupportedFeature, Battery]: + """Decipher CENTURION_BATTERY_SOC (0x0104) response. + + Response format (3 bytes): + Byte 0: Battery Percentage (0-100) + Byte 1: Battery Percentage (duplicate) + Byte 2: Charging Status (0=discharging, 1=charging, 2=charging via USB, 3=charge complete) + """ + if len(report) < 1: + return SupportedFeature.CENTURION_BATTERY_SOC, Battery(None, None, BatteryStatus.DISCHARGING, None) + soc = report[0] + logger.debug("centurion battery SOC raw: %s", report[:8].hex()) + charging_status = report[2] if len(report) >= 3 else 0 + if charging_status in (1, 2): + status = BatteryStatus.RECHARGING + elif charging_status == 3: + status = BatteryStatus.FULL + else: + status = BatteryStatus.DISCHARGING + return SupportedFeature.CENTURION_BATTERY_SOC, Battery(soc, None, status, None) + + +# --- CenturionReceiver class --- + + +class CenturionReceiver: + """A lightweight receiver-like container for Centurion (PRO X 2 LIGHTSPEED) dongles. + + Provides the Receiver interface to the UI so the dongle appears as a parent + with the headset as an indented child device. NOT a subclass of Receiver — + Receiver's __init__ does HID++ 1.0 register reads and pairing setup that + don't apply to Centurion. + + All centurion communication (bridge, features, settings, battery) lives in + the child Device; this class is just a UI container + handle owner. + """ + + read_register: Callable = hidpp10.read_register + write_register: Callable = hidpp10.write_register + number = 0xFF + kind = None + isDevice = False + may_unpair = False + re_pairs = False + max_devices = 1 + + def __init__(self, low_level, handle, device_info, setting_callback=None): + assert handle + self.low_level = low_level + self.handle = handle + self.path = device_info.path + self.product_id = device_info.product_id + self.setting_callback = setting_callback + self.status_callback = None + self.notification_flags = None + self._devices = {} + self._firmware = None + self._dongle_features = None # independently probed dongle features + self.cleanups = [] + + # Receiver identity + self.serial = None + self._usb_name = getattr(device_info, "product", None) + if not self._usb_name and self.path: + self._usb_name = _read_usb_product_string(self.path) + self.name = "Centurion Receiver" + + # Dummy pairing object — lock_open stays False + from .receiver import Pairing + + self.pairing = Pairing() + + # Discover dongle features independently + self._discover_dongle_features() + + # Read serial from dongle's CENTURION_DEVICE_INFO if available + if self.serial is None: + try: + s = get_serial_centurion(self) + if s and s.strip() and s.strip().isprintable(): + self.serial = s.strip() + except Exception: + pass + + def enable_connection_notifications(self, enable=True): + return False + + def remaining_pairings(self, cache=True): + return None + + def device_codename(self, n): + return self._usb_name + + def request(self, request_id, *params, no_reply=False): + """Send an HID++ request directly to the dongle (not through bridge).""" + if self.handle: + return self.low_level.request( + self.handle, 0xFF, request_id, *params, no_reply=no_reply, long_message=True, protocol=2.0 + ) + + def feature_request(self, feature, function=0x00, *params, no_reply=False): + """Send a feature request to the dongle using discovered feature indices.""" + if self._dongle_features is None: + self._discover_dongle_features() + feature_int = int(feature) + for _feat, feat_id, index in self._dongle_features or []: + if feat_id == feature_int: + request_id = (index << 8) | (function & 0xFF) + return self.request(request_id, *params, no_reply=no_reply) + raise exceptions.FeatureNotSupported(feature) + + def _discover_dongle_features(self): + """Independently discover features on the dongle hardware.""" + self._dongle_features = [] + try: + # Query ROOT for FEATURE_SET index + response = self.request(0x0000, 0x00, 0x01) + if response is None or response[0] == 0: + return + fs_index = response[0] + # Get feature count + count_resp = self.request(fs_index << 8) + if count_resp is None: + return + feature_count = count_resp[0] + # Enumerate features via CenturionFeatureSet (func 1 = 0x10, per-index query) + for idx in range(feature_count): + resp = self.request((fs_index << 8) | 0x10, idx) + if resp is None or len(resp) < 3: + continue + feat_id = struct.unpack("!H", resp[1:3])[0] + try: + feature = SupportedFeature(feat_id) + except ValueError: + feature = f"unknown:{feat_id:04X}" + self._dongle_features.append((feature, feat_id, idx)) + if logger.isEnabledFor(logging.DEBUG): + logger.debug("Centurion dongle features: %s", self._dongle_features) + except Exception: + logger.debug("Centurion dongle feature discovery failed", exc_info=True) + + @property + def dongle_features(self): + """Return list of (feature, feat_id, index) tuples for dongle features.""" + if self._dongle_features is None: + self._discover_dongle_features() + return self._dongle_features + + def count(self): + return len([d for d in self._devices.values() if d is not None]) + + @property + def firmware(self): + if self._firmware is None and self.handle: + self._firmware = get_firmware_centurion(self) + return self._firmware or () + + def notify_devices(self): + """Create child Device for the headset and trigger its initialization.""" + # Import Device locally to avoid circular import (centurion.py ↔ device.py) + from .device import Device + + # Signal receiver to UI first — tray/window need the receiver entry + # before a child device can be added under it. + self.changed(alert=Alert.NONE) + + # Create child Device with receiver=self, number=1 + pairing_info = { + "wpid": self.product_id, + "kind": None, + "serial": None, + "polling": None, + "power_switch": None, + } + dev = Device( + self.low_level, + self, + 1, + None, + pairing_info=pairing_info, + setting_callback=self.setting_callback, + ) + # Set centurion attributes on the child + dev.centurion = True + dev.product_id = self.product_id + dev.hidpp_long = True + dev._centurion_usb_name = self._usb_name + # Pre-set bridge index from dongle features so ping can probe the headset + for _feat, feat_id, idx in self._dongle_features or []: + if feat_id == 0x0003: # CentPPBridge + dev._centurion_bridge_index = idx + break + + self._devices[1] = dev + configuration.attach_to(dev) + dev.status_callback = self.status_callback + + # Ping to determine online status. + # Notify UI either way — offline devices show as greyed out (matching receiver behavior). + online = dev.ping() + dev.changed(active=online) + if self.status_callback is not None: + self.status_callback(dev) + + def changed(self, alert=Alert.NOTIFICATION, reason=None): + if self.status_callback is not None: + self.status_callback(self, alert=alert, reason=reason) + + def status_string(self): + count = self.count() + if count == 0: + return "No devices." + return f"{count} device connected." + + def close(self): + handle, self.handle = self.handle, None + for _n, d in self._devices.items(): + if d: + d.close() + self._devices.clear() + for cleanup in self.cleanups: + cleanup(self) + return handle and self.low_level.close(handle) + + def __del__(self): + self.close() + + def __iter__(self): + for dev in self._devices.values(): + if dev is not None: + yield dev + + def __getitem__(self, key): + dev = self._devices.get(key) + if dev is not None: + return dev + raise IndexError(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 __bool__(self): + return self.handle is not None + + __nonzero__ = __bool__ + + 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 "<%s(%s,%s%s)>" % ( + self.name.replace(" ", "") if self.name else "CenturionReceiver", + self.path, + "" if isinstance(self.handle, int) else "T", + self.handle, + ) + + __repr__ = __str__ + + +def create_centurion_receiver(low_level, device_info, setting_callback=None): + """Opens a Centurion dongle and wraps it as a receiver-like container. + + Creates a CenturionReceiver, discovers its features, then checks if + CentPPBridge (0x0003) is among them. If not, this is a direct-connected + device (wired headset) — close and return None so the caller can fall + back to create_device(). + + :returns: A CenturionReceiver, or None. + """ + try: + handle = low_level.open_path(device_info.path) + if handle: + base._centurion_handles.add(int(handle)) + cr = CenturionReceiver(low_level, handle, device_info, setting_callback) + # Check if any discovered feature is CentPPBridge (0x0003) + has_bridge = any(feat_id == 0x0003 for _, feat_id, _ in (cr.dongle_features or [])) + if not has_bridge: + logger.info("Centurion device %s has no bridge, treating as direct device", device_info.path) + base._centurion_handles.discard(int(handle)) + cr.handle = None # prevent __del__ from double-closing + low_level.close(handle) + return None + return cr + except OSError as e: + logger.exception("open %s", device_info) + if e.errno == errno.EACCES: + raise e + except Exception as e: + logger.exception("open %s", device_info) + raise e diff --git a/lib/logitech_receiver/common.py b/lib/logitech_receiver/common.py index 3d00a015..3aa4127e 100644 --- a/lib/logitech_receiver/common.py +++ b/lib/logitech_receiver/common.py @@ -674,3 +674,17 @@ class Notification(IntEnum): class BusID(IntEnum): USB = 0x03 BLUETOOTH = 0x05 + + +def _read_usb_product_string(hidraw_path): + """Read the USB product string from sysfs for a hidraw device path.""" + import pathlib + + try: + # /sys/class/hidraw/hidrawN/device/../../product → USB device product string + hidraw_name = pathlib.Path(hidraw_path).name + product_path = pathlib.Path("/sys/class/hidraw") / hidraw_name / "device" / ".." / ".." / "product" + product = product_path.read_text().strip() + return product if product else None + except (OSError, ValueError): + return None diff --git a/lib/logitech_receiver/device.py b/lib/logitech_receiver/device.py index cd60ff69..45c4ee2d 100644 --- a/lib/logitech_receiver/device.py +++ b/lib/logitech_receiver/device.py @@ -40,6 +40,7 @@ from . import settings from . import settings_templates from .common import Alert from .common import Battery +from .common import _read_usb_product_string from .hidpp10_constants import NotificationFlag from .hidpp20_constants import SupportedFeature @@ -69,296 +70,6 @@ class LowLevelInterface(Protocol): ... -def _read_usb_product_string(hidraw_path): - """Read the USB product string from sysfs for a hidraw device path.""" - import pathlib - - try: - # /sys/class/hidraw/hidrawN/device/../../product → USB device product string - hidraw_name = pathlib.Path(hidraw_path).name - product_path = pathlib.Path("/sys/class/hidraw") / hidraw_name / "device" / ".." / ".." / "product" - product = product_path.read_text().strip() - return product if product else None - except (OSError, ValueError): - return None - - -class CenturionReceiver: - """A lightweight receiver-like container for Centurion (PRO X 2 LIGHTSPEED) dongles. - - Provides the Receiver interface to the UI so the dongle appears as a parent - with the headset as an indented child device. NOT a subclass of Receiver — - Receiver's __init__ does HID++ 1.0 register reads and pairing setup that - don't apply to Centurion. - - All centurion communication (bridge, features, settings, battery) lives in - the child Device; this class is just a UI container + handle owner. - """ - - read_register: Callable = hidpp10.read_register - write_register: Callable = hidpp10.write_register - number = 0xFF - kind = None - isDevice = False - may_unpair = False - re_pairs = False - max_devices = 1 - - def __init__(self, low_level, handle, device_info, setting_callback=None): - assert handle - self.low_level = low_level - self.handle = handle - self.path = device_info.path - self.product_id = device_info.product_id - self.setting_callback = setting_callback - self.status_callback = None - self.notification_flags = None - self._devices = {} - self._firmware = None - self._dongle_features = None # independently probed dongle features - self.cleanups = [] - - # Receiver identity - self.serial = None - self._usb_name = getattr(device_info, "product", None) - if not self._usb_name and self.path: - self._usb_name = _read_usb_product_string(self.path) - self.name = "Centurion Receiver" - - # Dummy pairing object — lock_open stays False - from .receiver import Pairing - - self.pairing = Pairing() - - # Discover dongle features independently - self._discover_dongle_features() - - # Read serial from dongle's CENTURION_DEVICE_INFO if available - if self.serial is None: - try: - s = _hidpp20.get_serial_centurion(self) - if s and s.strip() and s.strip().isprintable(): - self.serial = s.strip() - except Exception: - pass - - def enable_connection_notifications(self, enable=True): - return False - - def remaining_pairings(self, cache=True): - return None - - def device_codename(self, n): - return self._usb_name - - def request(self, request_id, *params, no_reply=False): - """Send an HID++ request directly to the dongle (not through bridge).""" - if self.handle: - return self.low_level.request( - self.handle, 0xFF, request_id, *params, no_reply=no_reply, long_message=True, protocol=2.0 - ) - - def feature_request(self, feature, function=0x00, *params, no_reply=False): - """Send a feature request to the dongle using discovered feature indices.""" - if self._dongle_features is None: - self._discover_dongle_features() - feature_int = int(feature) - for _feat, feat_id, index in self._dongle_features or []: - if feat_id == feature_int: - request_id = (index << 8) | (function & 0xFF) - return self.request(request_id, *params, no_reply=no_reply) - raise exceptions.FeatureNotSupported(feature) - - def _discover_dongle_features(self): - """Independently discover features on the dongle hardware.""" - self._dongle_features = [] - try: - # Query ROOT for FEATURE_SET index - response = self.request(0x0000, 0x00, 0x01) - if response is None or response[0] == 0: - return - fs_index = response[0] - # Get feature count - count_resp = self.request(fs_index << 8) - if count_resp is None: - return - feature_count = count_resp[0] - # Enumerate features via CenturionFeatureSet (func 1 = 0x10, per-index query) - for idx in range(feature_count): - resp = self.request((fs_index << 8) | 0x10, idx) - if resp is None or len(resp) < 3: - continue - feat_id = struct.unpack("!H", resp[1:3])[0] - try: - feature = SupportedFeature(feat_id) - except ValueError: - feature = f"unknown:{feat_id:04X}" - self._dongle_features.append((feature, feat_id, idx)) - if logger.isEnabledFor(logging.DEBUG): - logger.debug("Centurion dongle features: %s", self._dongle_features) - except Exception: - logger.debug("Centurion dongle feature discovery failed", exc_info=True) - - @property - def dongle_features(self): - """Return list of (feature, feat_id, index) tuples for dongle features.""" - if self._dongle_features is None: - self._discover_dongle_features() - return self._dongle_features - - def count(self): - return len([d for d in self._devices.values() if d is not None]) - - @property - def firmware(self): - if self._firmware is None and self.handle: - self._firmware = _hidpp20.get_firmware_centurion(self) - return self._firmware or () - - def notify_devices(self): - """Create child Device for the headset and trigger its initialization.""" - # Signal receiver to UI first — tray/window need the receiver entry - # before a child device can be added under it. - self.changed(alert=Alert.NONE) - - # Create child Device with receiver=self, number=1 - pairing_info = { - "wpid": self.product_id, - "kind": None, - "serial": None, - "polling": None, - "power_switch": None, - } - dev = Device( - self.low_level, - self, - 1, - None, - pairing_info=pairing_info, - setting_callback=self.setting_callback, - ) - # Set centurion attributes on the child - dev.centurion = True - dev.product_id = self.product_id - dev.hidpp_long = True - dev._centurion_usb_name = self._usb_name - # Pre-set bridge index from dongle features so ping can probe the headset - for _feat, feat_id, idx in self._dongle_features or []: - if feat_id == 0x0003: # CentPPBridge - dev._centurion_bridge_index = idx - break - - self._devices[1] = dev - configuration.attach_to(dev) - dev.status_callback = self.status_callback - - # Ping to determine online status. - # Notify UI either way — offline devices show as greyed out (matching receiver behavior). - online = dev.ping() - dev.changed(active=online) - if self.status_callback is not None: - self.status_callback(dev) - - def changed(self, alert=Alert.NOTIFICATION, reason=None): - if self.status_callback is not None: - self.status_callback(self, alert=alert, reason=reason) - - def status_string(self): - count = self.count() - if count == 0: - return "No devices." - return f"{count} device connected." - - def close(self): - handle, self.handle = self.handle, None - for _n, d in self._devices.items(): - if d: - d.close() - self._devices.clear() - for cleanup in self.cleanups: - cleanup(self) - return handle and self.low_level.close(handle) - - def __del__(self): - self.close() - - def __iter__(self): - for dev in self._devices.values(): - if dev is not None: - yield dev - - def __getitem__(self, key): - dev = self._devices.get(key) - if dev is not None: - return dev - raise IndexError(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 __bool__(self): - return self.handle is not None - - __nonzero__ = __bool__ - - 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 "<%s(%s,%s%s)>" % ( - self.name.replace(" ", "") if self.name else "CenturionReceiver", - self.path, - "" if isinstance(self.handle, int) else "T", - self.handle, - ) - - __repr__ = __str__ - - -def create_centurion_receiver(low_level: LowLevelInterface, device_info, setting_callback=None): - """Opens a Centurion dongle and wraps it as a receiver-like container. - - Creates a CenturionReceiver, discovers its features, then checks if - CentPPBridge (0x0003) is among them. If not, this is a direct-connected - device (wired headset) — close and return None so the caller can fall - back to create_device(). - - :returns: A CenturionReceiver, or None. - """ - try: - handle = low_level.open_path(device_info.path) - if handle: - base._centurion_handles.add(int(handle)) - cr = CenturionReceiver(low_level, handle, device_info, setting_callback) - # Check if any discovered feature is CentPPBridge (0x0003) - has_bridge = any(feat_id == 0x0003 for _, feat_id, _ in (cr.dongle_features or [])) - if not has_bridge: - logger.info("Centurion device %s has no bridge, treating as direct device", device_info.path) - base._centurion_handles.discard(int(handle)) - cr.handle = None # prevent __del__ from double-closing - low_level.close(handle) - return None - return cr - except OSError as e: - logger.exception("open %s", device_info) - if e.errno == errno.EACCES: - raise e - except Exception as e: - logger.exception("open %s", device_info) - raise e - - def create_device(low_level: LowLevelInterface, device_info, setting_callback=None): """Opens a Logitech Device found attached to the machine, by Linux device path. :returns: An open file handle for the found receiver, or None. @@ -1186,3 +897,8 @@ class Device: def __del__(self): self.close() + + +# Re-export from centurion.py — must be after Device class to avoid circular import +from .centurion import CenturionReceiver # noqa: E402,F401 +from .centurion import create_centurion_receiver # noqa: E402,F401 diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index c07d076d..c9b2e638 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -17,7 +17,6 @@ from __future__ import annotations import logging -import math import socket import struct import threading @@ -36,6 +35,7 @@ import yaml from solaar.i18n import _ from typing_extensions import Protocol +from . import centurion as _centurion from . import common from . import exceptions from . import hidpp10_constants @@ -1687,106 +1687,25 @@ class Hidpp20: return tuple(fw) def get_firmware_centurion(self, device): - """Reads firmware info from a Centurion device via DeviceInfo (0x0100) function 1.""" - fw = [] - seen = set() # track response signatures to detect duplicates - for index in range(0, 8): # try up to 8 entities - try: - report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO, 0x10, index) - except exceptions.FeatureCallError: - break - if not report or len(report) < 5: - break - # Dedup: parent device returns the same response for every entity index - sig = bytes(report[: 5 + report[4]]) - if sig in seen: - break - seen.add(sig) - fw_type = report[0] - version = struct.unpack("!H", report[2:4])[0] - name_len = report[4] - name = report[5 : 5 + name_len].decode("ascii", errors="replace").rstrip("\x00") if name_len else "" - version_str = f"{version >> 8}.{version & 0xFF:02d}" - kind = FirmwareKind(fw_type) if fw_type <= 3 else FirmwareKind.Other - fw.append(common.FirmwareInfo(kind, name, version_str, None)) - return tuple(fw) if fw else None + return _centurion.get_firmware_centurion(device) def get_serial_centurion(self, device): - """Reads the serial number from a Centurion device via DeviceInfo (0x0100) function 2.""" - try: - report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO, 0x20) - except exceptions.FeatureCallError: - return None - if not report or len(report) < 2: - return None - str_len = report[0] - return report[1 : 1 + str_len].decode("ascii", errors="replace").rstrip("\x00") + return _centurion.get_serial_centurion(device) def get_hardware_info_centurion(self, device): - """Reads hardware info from a Centurion device via DeviceInfo (0x0100) function 0. - - Returns (modelId, hardwareRevision, productId) or None. - """ - try: - report = device.feature_request(SupportedFeature.CENTURION_DEVICE_INFO) - except exceptions.FeatureCallError: - return None - if not report or len(report) < 4: - return None - model_id = report[0] - hw_revision = report[1] - product_id = struct.unpack("!H", report[2:4])[0] - return model_id, hw_revision, product_id + return _centurion.get_hardware_info_centurion(device) def _centurion_sub_device_info_request(self, device, function=0x00, *params): - """Send a DeviceInfo (0x0100) request to the sub-device via bridge.""" - sub_indices = getattr(device, "_centurion_sub_indices", {}) - sub_idx = sub_indices.get(SupportedFeature.CENTURION_DEVICE_INFO) - if sub_idx is None: - return None - return device.centurion_bridge_request(sub_idx, function, *params) + return _centurion._centurion_sub_device_info_request(device, function, *params) def get_firmware_centurion_sub(self, device): - """Reads firmware info from the Centurion sub-device (headset) via bridge.""" - fw = [] - seen = set() - for index in range(0, 8): - report = self._centurion_sub_device_info_request(device, 0x10, index) - if not report or len(report) < 5: - break - sig = bytes(report[: 5 + report[4]]) - if sig in seen: - break - seen.add(sig) - fw_type = report[0] - version = struct.unpack("!H", report[2:4])[0] - name_len = report[4] - name = report[5 : 5 + name_len].decode("ascii", errors="replace").rstrip("\x00") if name_len else "" - version_str = f"{version >> 8}.{version & 0xFF:02d}" - kind = FirmwareKind(fw_type) if fw_type <= 3 else FirmwareKind.Other - fw.append(common.FirmwareInfo(kind, name, version_str, None)) - return tuple(fw) if fw else None + return _centurion.get_firmware_centurion_sub(device) def get_serial_centurion_sub(self, device): - """Reads the serial number from the Centurion sub-device (headset) via bridge.""" - report = self._centurion_sub_device_info_request(device, 0x20) - if not report or len(report) < 2: - return None - str_len = report[0] - return report[1 : 1 + str_len].decode("ascii", errors="replace").rstrip("\x00") + return _centurion.get_serial_centurion_sub(device) def get_hardware_info_centurion_sub(self, device): - """Reads hardware info from the Centurion sub-device (headset) via bridge. - - Returns (modelId, hardwareRevision, productId) or None. - """ - report = self._centurion_sub_device_info_request(device) - if not report or len(report) < 4: - return None - model_id = report[0] - hw_revision = report[1] - product_id = struct.unpack("!H", report[2:4])[0] - return model_id, hw_revision, product_id + return _centurion.get_hardware_info_centurion_sub(device) def get_ids(self, device): """Reads a device's ids (unit and model numbers)""" @@ -1840,36 +1759,7 @@ class Hidpp20: return name.decode("utf-8") def get_name_centurion(self, device): - """Reads a Centurion device's name via DeviceName (0x0101). - - Tries two response formats: - 1. Inline: function 0 returns [name_len, name_bytes...] (like serial) - 2. Chunked: function 0 returns [name_len], function 1 returns [name_bytes...] (like standard DeviceName) - """ - try: - reply = device.feature_request(SupportedFeature.CENTURION_DEVICE_NAME) - except exceptions.FeatureCallError: - return None - if not reply: - return None - name_length = reply[0] - if name_length == 0: - return None - # If the full name is inline (length + name bytes in one response) - if len(reply) >= 1 + name_length: - return reply[1 : 1 + name_length].decode("utf-8", errors="replace").rstrip("\x00") - # Otherwise, fetch name in chunks via function 1 (like standard DEVICE_NAME) - name = b"" - while len(name) < name_length: - try: - fragment = device.feature_request(SupportedFeature.CENTURION_DEVICE_NAME, 0x10, len(name)) - except exceptions.FeatureCallError: - break - if fragment: - name += fragment[: name_length - len(name)] - else: - break - return name.decode("utf-8", errors="replace").rstrip("\x00") if name else None + return _centurion.get_name_centurion(device) def get_friendly_name(self, device: Device): """Reads a device's friendly name. @@ -1916,14 +1806,7 @@ class Hidpp20: return SupportedFeature.ADC_MEASUREMENT if SupportedFeature.ADC_MEASUREMENT in device.features else None def get_battery_centurion(self, device: Device): - try: - report = device.feature_request(SupportedFeature.CENTURION_BATTERY_SOC) - if report is not None: - return decipher_battery_centurion(report) - except exceptions.FeatureCallError: - if SupportedFeature.CENTURION_BATTERY_SOC in device.features: - return SupportedFeature.CENTURION_BATTERY_SOC - return None + return _centurion.get_battery_centurion(device) def get_battery(self, device, feature): """Return battery information - feature, approximate level, next, charging, voltage @@ -2237,26 +2120,7 @@ def decipher_battery_unified(report) -> tuple[SupportedFeature, Battery]: return SupportedFeature.UNIFIED_BATTERY, Battery(discharge if discharge else approx_level, None, status, None) -def decipher_battery_centurion(report) -> tuple[SupportedFeature, Battery]: - """Decipher CENTURION_BATTERY_SOC (0x0104) response. - - Response format (3 bytes): - Byte 0: Battery Percentage (0-100) - Byte 1: Battery Percentage (duplicate) - Byte 2: Charging Status (0=discharging, 1=charging, 2=charging via USB, 3=charge complete) - """ - if len(report) < 1: - return SupportedFeature.CENTURION_BATTERY_SOC, Battery(None, None, BatteryStatus.DISCHARGING, None) - soc = report[0] - logger.debug("centurion battery SOC raw: %s", report[:8].hex()) - charging_status = report[2] if len(report) >= 3 else 0 - if charging_status in (1, 2): - status = BatteryStatus.RECHARGING - elif charging_status == 3: - status = BatteryStatus.FULL - else: - status = BatteryStatus.DISCHARGING - return SupportedFeature.CENTURION_BATTERY_SOC, Battery(soc, None, status, None) +decipher_battery_centurion = _centurion.decipher_battery_centurion def decipher_adc_measurement(report) -> tuple[SupportedFeature, Battery]: @@ -2400,162 +2264,8 @@ class ForceSensingButtonArray(UserDict): return self[index].acceptable(value) -# --- OnboardEQ (0x0636) biquad coefficient math and helpers --- - -# Mystery bytes observed in every LGHUB pcap EQ write between band params -# and coefficient header. Purpose not fully understood — possibly a null-band -# terminator for DSPs that support >5 bands (advanced 10-band mode). -# First byte matches band_count; bytes 2-3 look like LE16 coeff blob size. -# Hardcoded from pcap for initial bring-up; revisit once device-tested. -_EQ_MYSTERY_BYTES = b"\x05\x5a\xe3\x00" - - -def _peaking_eq_biquad(freq_hz, gain_db, Q, sample_rate=48000.0): - """Compute peaking EQ biquad coefficients (Audio EQ Cookbook). - - Returns (b0/a0, b1/a0, b2/a0, a1/a0, a2/a0) normalised coefficients. - """ - A = 10.0 ** (gain_db / 40.0) - w0 = 2.0 * math.pi * freq_hz / sample_rate - cos_w0 = math.cos(w0) - alpha = math.sin(w0) / (2.0 * Q) - a0 = 1.0 + alpha / A - return ( - (1.0 + alpha * A) / a0, - (-2.0 * cos_w0) / a0, - (1.0 - alpha * A) / a0, - (-2.0 * cos_w0) / a0, - (1.0 - alpha / A) / a0, - ) - - -def _quantize_coeffs(b0, b1, b2, a1, a2): - """Quantize biquad coefficients to mixed Q1.31 / Q2.30 fixed-point. - - b0, b2, a2 use Q1.31 (x 2^31); b1, a1 use Q2.30 (x 2^30). - Values are truncated to 24-bit precision (low byte zeroed) matching - the device DSP's internal format. - Returns list of 10 uint16 values (5 coefficients x 2 LE words each, - high word first). - """ - scales = [2**31, 2**30, 2**31, 2**30, 2**31] # b0, b1, b2, a1, a2 - words = [] - for val, scale in zip([b0, b1, b2, a1, a2], scales): - q = int(round(val * scale)) - q = max(-(1 << 31), min((1 << 31) - 1, q)) - q = q & 0xFFFFFF00 # 24-bit precision (low byte always zero) - words.append((q >> 16) & 0xFFFF) # high word - words.append(q & 0xFFFF) # low word - return words - - -def _build_coeff_section(bands, sample_rate, section_type=1): - """Build one coefficient section for a DSP processing block. - - Returns bytes: 4-byte section header + coefficient data as LE uint16 words. - Section header: [type, 0x00, count_lo, count_hi]. - - Coefficients are normalized by a rescale factor to prevent Q1.31 overflow. - Only feedforward coefficients (b0, b1, b2) are divided by rescale; feedback - coefficients (a1, a2) are left unchanged. The DSP multiplies the output by - rescale to restore correct gain. - """ - _HEADROOM = 1.19 # 19% headroom margin (matches LGHUB) - num_bands = len(bands) - all_words = [num_bands] # first uint16 = num_bands - - # First pass: compute raw biquad coefficients for all bands - raw_coeffs = [] - for freq, gain, Q in bands: - raw_coeffs.append(_peaking_eq_biquad(freq, gain, max(Q, 0.1), sample_rate)) - - # Compute rescale: ensure max |b0| fits in Q1.31 with headroom - max_b0 = max(abs(c[0]) for c in raw_coeffs) - rescale = max(1.0, max_b0) * _HEADROOM - - # Second pass: normalize b-coefficients and quantize - for b0, b1, b2, a1, a2 in raw_coeffs: - all_words.extend(_quantize_coeffs(b0 / rescale, b1 / rescale, b2 / rescale, a1, a2)) - - # Rescale factor as Q6.26, 24-bit precision - rs = int(round(rescale * (1 << 26))) - rs = max(-(1 << 31), min((1 << 31) - 1, rs)) & 0xFFFFFF00 - all_words.append((rs >> 16) & 0xFFFF) - all_words.append(rs & 0xFFFF) - - coeff_count = num_bands * 10 + 3 # num_bands word + 10 per band + 2 rescale words - hdr = bytes([section_type, 0x00, coeff_count & 0xFF, (coeff_count >> 8) & 0xFF]) - data = struct.pack(f"<{len(all_words)}H", *all_words) - return hdr + data - - -def _build_eq_coeffs_payload(bands): - """Build the full EQCoeffs wire payload for SetEQParameters. - - Two coefficient sections: type=1 (48 kHz playback) and type=2 (16 kHz mic). - Returns bytes: 7-byte header + sections (no trailing padding). - """ - section_count = 2 - header = bytes([0x03, 0x0E, 0x00, section_count, 0x00, 0x00, 0x00]) - sections = _build_coeff_section(bands, 48000.0, section_type=1) - sections += _build_coeff_section(bands, 16000.0, section_type=2) - return header + sections - - -def _build_set_eq_payload(slot, bands): - """Build complete SetEQParameters payload: band params + biquad coefficients. - - bands: list of (freq_hz, gain_db, Q) tuples. - Returns bytes ready to send as sub-device params. - """ - params = bytes([slot, len(bands)]) - for freq, gain, Q in bands: - params += struct.pack(">H", freq) + bytes([gain & 0xFF, Q & 0xFF]) - params += _EQ_MYSTERY_BYTES - params += _build_eq_coeffs_payload(bands) - return params - - -def get_onboard_eq_info(device): - """Query HEADSET_ONBOARD_EQ GetEQInfos (function 0). - - Returns (has_hw_eq, num_bands) or None. - """ - result = device.feature_request(SupportedFeature.HEADSET_ONBOARD_EQ, 0x00) - if result is None or len(result) < 5: - return None - has_hw_eq = bool(result[0] & 0x80) - num_bands = result[4] - return (has_hw_eq, num_bands) - - -def get_onboard_eq_params(device, slot=0x00): - """Query HEADSET_ONBOARD_EQ GetEQParameters (function 0x10). - - Returns list of (freq_hz, gain_db, q) tuples, or None. - """ - result = device.feature_request(SupportedFeature.HEADSET_ONBOARD_EQ, 0x10, slot) - if result is None or len(result) < 2: - return None - band_count = result[1] - bands = [] - offset = 2 - for _i in range(band_count): - if offset + 4 > len(result): - break - freq_hz = struct.unpack(">H", result[offset : offset + 2])[0] - gain_db = struct.unpack("b", bytes([result[offset + 2]]))[0] # signed - q = result[offset + 3] - bands.append((freq_hz, gain_db, q)) - offset += 4 - return bands - - -def set_onboard_eq_params(device, bands, slot=0x00): - """Send HEADSET_ONBOARD_EQ SetEQParameters (function 0x20). - - bands: list of (freq_hz, gain_db, Q) tuples. - Returns response or None. - """ - payload = _build_set_eq_payload(slot, bands) - return device.feature_request(SupportedFeature.HEADSET_ONBOARD_EQ, 0x20, payload) +# --- OnboardEQ (0x0636) — re-exported from onboard_eq.py --- +from .onboard_eq import _build_set_eq_payload # noqa: E402, F401 +from .onboard_eq import get_onboard_eq_info # noqa: E402, F401 +from .onboard_eq import get_onboard_eq_params # noqa: E402, F401 +from .onboard_eq import set_onboard_eq_params # noqa: E402, F401 diff --git a/lib/logitech_receiver/onboard_eq.py b/lib/logitech_receiver/onboard_eq.py new file mode 100644 index 00000000..107b11f6 --- /dev/null +++ b/lib/logitech_receiver/onboard_eq.py @@ -0,0 +1,186 @@ +## Copyright (C) 2012-2013 Daniel Pavel +## Copyright (C) 2014-2024 Solaar Contributors https://pwr-solaar.github.io/Solaar/ +## +## 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. + +"""OnboardEQ (0x0636) biquad coefficient math and payload builders. + +Pure computation — no device or transport dependencies beyond feature_request(). +""" + +from __future__ import annotations + +import math +import struct + +from .hidpp20_constants import SupportedFeature + +# Mystery bytes observed in every LGHUB pcap EQ write between band params +# and coefficient header. Purpose not fully understood — possibly a null-band +# terminator for DSPs that support >5 bands (advanced 10-band mode). +# First byte matches band_count; bytes 2-3 look like LE16 coeff blob size. +# Hardcoded from pcap for initial bring-up; revisit once device-tested. +_EQ_MYSTERY_BYTES = b"\x05\x5a\xe3\x00" + + +def _peaking_eq_biquad(freq_hz, gain_db, Q, sample_rate=48000.0): + """Compute peaking EQ biquad coefficients (Audio EQ Cookbook). + + Returns (b0/a0, b1/a0, b2/a0, a1/a0, a2/a0) normalised coefficients. + """ + A = 10.0 ** (gain_db / 40.0) + w0 = 2.0 * math.pi * freq_hz / sample_rate + cos_w0 = math.cos(w0) + alpha = math.sin(w0) / (2.0 * Q) + a0 = 1.0 + alpha / A + return ( + (1.0 + alpha * A) / a0, + (-2.0 * cos_w0) / a0, + (1.0 - alpha * A) / a0, + (-2.0 * cos_w0) / a0, + (1.0 - alpha / A) / a0, + ) + + +def _quantize_coeffs(b0, b1, b2, a1, a2): + """Quantize biquad coefficients to mixed Q1.31 / Q2.30 fixed-point. + + b0, b2, a2 use Q1.31 (x 2^31); b1, a1 use Q2.30 (x 2^30). + Values are truncated to 24-bit precision (low byte zeroed) matching + the device DSP's internal format. + Returns list of 10 uint16 values (5 coefficients x 2 LE words each, + high word first). + """ + scales = [2**31, 2**30, 2**31, 2**30, 2**31] # b0, b1, b2, a1, a2 + words = [] + for val, scale in zip([b0, b1, b2, a1, a2], scales): + q = int(round(val * scale)) + q = max(-(1 << 31), min((1 << 31) - 1, q)) + q = q & 0xFFFFFF00 # 24-bit precision (low byte always zero) + words.append((q >> 16) & 0xFFFF) # high word + words.append(q & 0xFFFF) # low word + return words + + +def _build_coeff_section(bands, sample_rate, section_type=1): + """Build one coefficient section for a DSP processing block. + + Returns bytes: 4-byte section header + coefficient data as LE uint16 words. + Section header: [type, 0x00, count_lo, count_hi]. + + Coefficients are normalized by a rescale factor to prevent Q1.31 overflow. + Only feedforward coefficients (b0, b1, b2) are divided by rescale; feedback + coefficients (a1, a2) are left unchanged. The DSP multiplies the output by + rescale to restore correct gain. + """ + _HEADROOM = 1.19 # 19% headroom margin (matches LGHUB) + num_bands = len(bands) + all_words = [num_bands] # first uint16 = num_bands + + # First pass: compute raw biquad coefficients for all bands + raw_coeffs = [] + for freq, gain, Q in bands: + raw_coeffs.append(_peaking_eq_biquad(freq, gain, max(Q, 0.1), sample_rate)) + + # Compute rescale: ensure max |b0| fits in Q1.31 with headroom + max_b0 = max(abs(c[0]) for c in raw_coeffs) + rescale = max(1.0, max_b0) * _HEADROOM + + # Second pass: normalize b-coefficients and quantize + for b0, b1, b2, a1, a2 in raw_coeffs: + all_words.extend(_quantize_coeffs(b0 / rescale, b1 / rescale, b2 / rescale, a1, a2)) + + # Rescale factor as Q6.26, 24-bit precision + rs = int(round(rescale * (1 << 26))) + rs = max(-(1 << 31), min((1 << 31) - 1, rs)) & 0xFFFFFF00 + all_words.append((rs >> 16) & 0xFFFF) + all_words.append(rs & 0xFFFF) + + coeff_count = num_bands * 10 + 3 # num_bands word + 10 per band + 2 rescale words + hdr = bytes([section_type, 0x00, coeff_count & 0xFF, (coeff_count >> 8) & 0xFF]) + data = struct.pack(f"<{len(all_words)}H", *all_words) + return hdr + data + + +def _build_eq_coeffs_payload(bands): + """Build the full EQCoeffs wire payload for SetEQParameters. + + Two coefficient sections: type=1 (48 kHz playback) and type=2 (16 kHz mic). + Returns bytes: 7-byte header + sections (no trailing padding). + """ + section_count = 2 + header = bytes([0x03, 0x0E, 0x00, section_count, 0x00, 0x00, 0x00]) + sections = _build_coeff_section(bands, 48000.0, section_type=1) + sections += _build_coeff_section(bands, 16000.0, section_type=2) + return header + sections + + +def _build_set_eq_payload(slot, bands): + """Build complete SetEQParameters payload: band params + biquad coefficients. + + bands: list of (freq_hz, gain_db, Q) tuples. + Returns bytes ready to send as sub-device params. + """ + params = bytes([slot, len(bands)]) + for freq, gain, Q in bands: + params += struct.pack(">H", freq) + bytes([gain & 0xFF, Q & 0xFF]) + params += _EQ_MYSTERY_BYTES + params += _build_eq_coeffs_payload(bands) + return params + + +def get_onboard_eq_info(device): + """Query HEADSET_ONBOARD_EQ GetEQInfos (function 0). + + Returns (has_hw_eq, num_bands) or None. + """ + result = device.feature_request(SupportedFeature.HEADSET_ONBOARD_EQ, 0x00) + if result is None or len(result) < 5: + return None + has_hw_eq = bool(result[0] & 0x80) + num_bands = result[4] + return (has_hw_eq, num_bands) + + +def get_onboard_eq_params(device, slot=0x00): + """Query HEADSET_ONBOARD_EQ GetEQParameters (function 0x10). + + Returns list of (freq_hz, gain_db, q) tuples, or None. + """ + result = device.feature_request(SupportedFeature.HEADSET_ONBOARD_EQ, 0x10, slot) + if result is None or len(result) < 2: + return None + band_count = result[1] + bands = [] + offset = 2 + for _i in range(band_count): + if offset + 4 > len(result): + break + freq_hz = struct.unpack(">H", result[offset : offset + 2])[0] + gain_db = struct.unpack("b", bytes([result[offset + 2]]))[0] # signed + q = result[offset + 3] + bands.append((freq_hz, gain_db, q)) + offset += 4 + return bands + + +def set_onboard_eq_params(device, bands, slot=0x00): + """Send HEADSET_ONBOARD_EQ SetEQParameters (function 0x20). + + bands: list of (freq_hz, gain_db, Q) tuples. + Returns response or None. + """ + payload = _build_set_eq_payload(slot, bands) + return device.feature_request(SupportedFeature.HEADSET_ONBOARD_EQ, 0x20, payload)