Extract Centurion protocol into separate modules
Move CenturionReceiver class, factory function, and Centurion protocol queries (firmware, serial, hardware info, battery, name) from device.py and hidpp20.py into new centurion.py module. Move OnboardEQ biquad math and payload builders from hidpp20.py into new onboard_eq.py module. Move _read_usb_product_string() to common.py to avoid circular imports. Re-exports preserve backward compatibility for all existing callers.
This commit is contained in:
Ken Sanislo
parent
14807fdf28
commit
b439c2d53f
|
|
@ -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
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue