RomanNum3ral
/
Solaar
mirror of https://github.com/pwr-Solaar/Solaar.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AdvancedParaEQ: add V2 wire-format support; keep V0/V1 

G522's 0x020D V2 uses a 5-byte band stride
[freq_hi, freq_lo, gain_i8, q_hi, q_lo] and a 13-byte getEQInfos
(gain bounds, gain_steps, format, xy, preset counts). Frequency and
Q are opaque u16 round-trip values — the u16->Hz and u16->Q mappings
need a LGHUB pcap to pin down (documented in
~/ghub/HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md).

get_advanced_eq_info now returns a dict with a `version` discriminator
and the union of V0/V1 and V2 fields; step_db is derived from the
gain_min/gain_max/gain_steps triple on V2 (0.05 dB/LSB on G522).

get_advanced_eq_params version-switches: V2 uses parse_v2_bands which
probes header length {5, 2, 0} until the tail is a clean multiple of
5, strips trailing all-zero terminator entries. V0/V1 falls through
to the legacy 3-byte stride so older devices still work.

HeadsetAdvancedEQ.build() no longer requires band_count from
getEQInfos (V2 doesn't advertise it); derives from getCustomEQ length
per the wire-protocol doc's recommendation. V2 band labels use
"Band N" since u16->Hz isn't confirmed. Read-only still — writes
stay gated until pcap confirms the encodings.
This commit is contained in:
Ken Sanislo 2026-04-19 11:30:32 -07:00
parent 1bcee309d3
commit 41db76bc81
3 changed files with 235 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

199
lib/logitech_receiver/advanced_para_eq.py
View File

@ -16,10 +16,18 @@
"""AdvancedParaEQ (0x020D) helpers.
Unlike OnboardEQ (0x0636) which requires host-computed biquad coefficients,
AdvancedParaEQ is handled entirely by the device — we send frequency + gain
per band and the device applies them. Band entries are 3 bytes each:
[freq_hi, freq_lo, value] where value is the signed int8 gain in dB.
The device handles biquad coefficient computation — we transmit only
per-band frequency, gain, and (on V2) Q-factor; the DSP does the rest.
V0/V1 wire format: 3-byte band stride [freq_hi, freq_lo, gain_i8];
getEQInfos returns 5 bytes [bandCount, dbRange, caps, dbMin, dbMax].
V2 wire format: 5-byte band stride [freq_hi, freq_lo, gain_i8, q_hi,
q_lo]; getEQInfos returns 13 bytes with gain bounds + step count,
format enum, XY-support flag, and onboard preset counts. Frequency and
Q are opaque u16 round-trip values — the u16→Hz / u16→Q mappings are
unconfirmed and need a LGHUB pcap to pin down. See
HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md.
"""
from __future__ import annotations
@ -31,39 +39,108 @@ from .hidpp20_constants import SupportedFeature
logger = logging.getLogger(__name__)
# Direction parameter for getCustomEQ / getActiveEQ etc.
DIRECTION_PLAYBACK = 0
DIRECTION_CAPTURE = 1
def get_advanced_eq_info(device):
"""Query HEADSET_ADVANCED_PARA_EQ getEQInfos (function 0).
def _get_version(device) -> int:
return device.features.get_feature_version(SupportedFeature.HEADSET_ADVANCED_PARA_EQ) or 0
Returns (band_count, db_range, capabilities, db_min, db_max) or None.
def get_advanced_eq_info(device):
"""Query getEQInfos (function 0). Returns a dict or None.
Common fields:
version int feature version (0, 1, 2)
gain_min_db int signed whole-dB min
gain_max_db int signed whole-dB max
step_db float dB per raw LSB (1.0 on V0/V1)
V0/V1 only:
band_count int number of bands (from wire byte 0)
db_range int raw byte 1
capabilities int raw byte 2
V2 only:
gain_steps int discrete gain positions
format int 0=CLASSIC, 1=STYLES
supports_xy bool
onboard_ro_preset_count int factory preset slots
onboard_custom_preset_count int user-writable preset slots
"""
version = _get_version(device)
result = device.feature_request(SupportedFeature.HEADSET_ADVANCED_PARA_EQ, 0x00)
if result is None:
logger.info("AdvancedParaEQ getEQInfos: feature_request returned None")
logger.info("AdvancedParaEQ getEQInfos V%d: feature_request returned None", version)
return None
if version >= 2:
if len(result) < 13:
logger.info("AdvancedParaEQ getEQInfos V2: short response len=%d %s", len(result), result.hex())
return None
gain_min = struct.unpack("b", bytes([result[2]]))[0]
gain_max = struct.unpack("b", bytes([result[3]]))[0]
gain_steps = struct.unpack(">H", result[4:6])[0]
fmt = result[6]
supports_xy = bool(result[7])
ro_presets = result[9]
custom_presets = result[10]
step_db = (gain_max - gain_min) / max(1, gain_steps - 1)
info = {
"version": 2,
"gain_min_db": gain_min,
"gain_max_db": gain_max,
"gain_steps": gain_steps,
"step_db": step_db,
"format": fmt,
"supports_xy": supports_xy,
"onboard_ro_preset_count": ro_presets,
"onboard_custom_preset_count": custom_presets,
}
logger.info(
"AdvancedParaEQ getEQInfos V2: gain=[%d,%d] steps=%d step_db=%.4f format=%d xy=%s "
"presets_ro=%d presets_custom=%d raw=%s",
gain_min,
gain_max,
gain_steps,
step_db,
fmt,
supports_xy,
ro_presets,
custom_presets,
result.hex(),
)
return info
# V0 / V1
if len(result) < 5:
logger.info("AdvancedParaEQ getEQInfos: short response (len=%d) %s", len(result), result.hex())
logger.info("AdvancedParaEQ getEQInfos V%d: short response len=%d %s", version, len(result), result.hex())
return None
band_count = result[0]
db_range = result[1]
capabilities = result[2]
# dbMin / dbMax are signed int8 in the doc.
db_min = struct.unpack("b", bytes([result[3]]))[0]
db_max = struct.unpack("b", bytes([result[4]]))[0]
caps = result[2]
gain_min = struct.unpack("b", bytes([result[3]]))[0]
gain_max = struct.unpack("b", bytes([result[4]]))[0]
info = {
"version": version,
"band_count": band_count,
"db_range": db_range,
"capabilities": caps,
"gain_min_db": gain_min,
"gain_max_db": gain_max,
"step_db": 1.0,
}
logger.info(
"AdvancedParaEQ getEQInfos: bands=%d dbRange=%d caps=0x%02X dbMin=%d dbMax=%d raw=%s",
"AdvancedParaEQ getEQInfos V%d: bands=%d dbRange=%d caps=0x%02X gain=[%d,%d] raw=%s",
version,
band_count,
db_range,
capabilities,
db_min,
db_max,
caps,
gain_min,
gain_max,
result.hex(),
)
return (band_count, db_range, capabilities, db_min, db_max)
return info
def get_advanced_eq_active_slot(device, direction=DIRECTION_PLAYBACK):
@ -79,29 +156,95 @@ def get_advanced_eq_active_slot(device, direction=DIRECTION_PLAYBACK):
return result[0]
def get_advanced_eq_params(device, direction=DIRECTION_PLAYBACK, slot=0):
"""Query getCustomEQ (function 1) for the given direction and slot.
def _parse_v2_band_payload(result: bytes):
"""Locate and parse the 5-byte band stride inside a V2 getCustomEQ response.
Returns a list of (freq_hz, gain_db) tuples, or None.
Header length before the first band is not yet nailed down (see
HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md section on band header).
Try candidate lengths {5, 2, 0} and pick the first where the tail is
a clean multiple of 5.
Returns (bands_bytes, header_len) or (None, None).
"""
for hl in (5, 2, 0):
tail = result[hl:]
if tail and len(tail) % 5 == 0 and 1 <= len(tail) // 5 <= 64:
return tail, hl
return None, None
def parse_v2_bands(result: bytes, step_db: float):
"""Parse a V2 getCustomEQ response. Returns list of (freq_u16, gain_db, q_u16).
Trailing all-zero bands (terminators) are stripped.
"""
payload, header_len = _parse_v2_band_payload(result)
if payload is None:
return None, None
bands = []
for i in range(len(payload) // 5):
e = payload[i * 5 : (i + 1) * 5]
freq_u16 = (e[0] << 8) | e[1]
gain_raw = struct.unpack("b", bytes([e[2]]))[0]
q_u16 = (e[3] << 8) | e[4]
bands.append((freq_u16, gain_raw * step_db, q_u16))
while bands and bands[-1] == (0, 0.0, 0):
bands.pop()
return bands, header_len
def get_advanced_eq_params(device, direction=DIRECTION_PLAYBACK, slot=0):
"""Query getCustomEQ (function 1). Returns list of (freq, gain_db, q) or None.
V0/V1: freq is raw Hz (u16), q is always 0 (V0/V1 has no Q).
V2: freq is opaque u16 bin index (Hz mapping unconfirmed), q is opaque u16
round-trip value (scale unconfirmed). See wire-protocol doc.
step_db for V2 is derived from getEQInfos; the caller should pass it via
`device._advanced_eq_info` (set by get_advanced_eq_info) or we fall back
to 1.0 and log a warning.
"""
version = _get_version(device)
result = device.feature_request(SupportedFeature.HEADSET_ADVANCED_PARA_EQ, 0x10, direction, slot)
if result is None:
logger.info("AdvancedParaEQ getCustomEQ(dir=%d slot=%d): feature_request returned None", direction, slot)
logger.info("AdvancedParaEQ getCustomEQ V%d (dir=%d slot=%d): feature_request returned None", version, direction, slot)
return None
if version >= 2:
info = getattr(device, "_advanced_eq_info", None)
step_db = info["step_db"] if info and "step_db" in info else 1.0
if step_db == 1.0 and not info:
logger.warning(
"AdvancedParaEQ getCustomEQ V2: no cached getEQInfos — gain values will use step_db=1.0 and be wrong"
)
bands, header_len = parse_v2_bands(result, step_db)
if bands is None:
logger.info("AdvancedParaEQ getCustomEQ V2: couldn't locate band payload raw=%s", result.hex())
return None
logger.info(
"AdvancedParaEQ getCustomEQ V2 (dir=%d slot=%d): parsed %d band(s) header_len=%d step_db=%.4f raw=%s",
direction,
slot,
len(bands),
header_len,
step_db,
result.hex(),
)
return bands
# V0 / V1: 3-byte stride, freq is raw Hz, gain is whole dB, no Q.
bands = []
offset = 0
# Response is a tight-packed series of 3-byte band entries:
# [freq_hi, freq_lo, value].
while offset + 3 <= len(result):
freq = struct.unpack(">H", result[offset : offset + 2])[0]
if freq == 0:
# Trailing padding — stop parsing.
break
gain_db = struct.unpack("b", bytes([result[offset + 2]]))[0]
bands.append((freq, gain_db))
bands.append((freq, float(gain_db), 0))
offset += 3
logger.info(
"AdvancedParaEQ getCustomEQ(dir=%d slot=%d): parsed %d band(s) %s raw=%s",
"AdvancedParaEQ getCustomEQ V%d (dir=%d slot=%d): parsed %d band(s) %s raw=%s",
version,
direction,
slot,
len(bands),

1
lib/logitech_receiver/hidpp20.py
View File

@ -2319,6 +2319,7 @@ class ForceSensingButtonArray(UserDict):
from .advanced_para_eq import get_advanced_eq_active_slot # noqa: E402, F401
from .advanced_para_eq import get_advanced_eq_info # noqa: E402, F401
from .advanced_para_eq import get_advanced_eq_params # noqa: E402, F401
from .advanced_para_eq import parse_v2_bands # noqa: E402, F401
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

86
lib/logitech_receiver/settings_templates.py
View File

@ -1855,15 +1855,15 @@ class HeadsetOnboardEQ(settings.RangeFieldSetting):
class HeadsetAdvancedEQ(settings.RangeFieldSetting):
"""Read-only display of the headset's active AdvancedParaEQ (0x020D) bands.
Writes are intentionally disabled for now — we want to verify the wire
format matches the protocol doc on real hardware before allowing any
changes that could misconfigure the device DSP. Once confirmed working
we'll wire up write() to call setCustomEQ.
Writes are intentionally disabled for now — V2's frequency and Q encodings
are still opaque u16 round-trip values (confirmed via LGHUB RE, see
HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md), so we can show the current
EQ but can't safely author a write until we have a LGHUB pcap that pins
down the u16→Hz and u16→Q mappings.
Bands come from getActiveEQ → getCustomEQ on the playback direction.
Each band entry is [freq_hi, freq_lo, gain_db] (3-byte stride). Unlike
OnboardEQ (0x0636), the device handles biquad coefficient computation
— we just send frequency and gain.
V0/V1: 3-byte band stride [freq_hi, freq_lo, gain_i8], gain is whole dB.
V2: 5-byte band stride [freq_hi, freq_lo, gain_i8, q_hi, q_lo], gain
is `signed_byte × step_db` where step_db comes from getEQInfos.
"""
name = "headset-advanced-eq"
@ -1882,40 +1882,80 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
if not info:
logger.info("HeadsetAdvancedEQ.build: getEQInfos failed, no panel will be built")
return None
band_count, _db_range, _caps, db_min, db_max = info
# Use the active slot on playback direction so the displayed EQ
# matches what the user is actually hearing.
# Cache so get_advanced_eq_params can look up step_db.
device._advanced_eq_info = info
version = info["version"]
gain_min = info["gain_min_db"]
gain_max = info["gain_max_db"]
step_db = info["step_db"]
active_slot = hidpp20.get_advanced_eq_active_slot(device, direction=0) or 0
bands = hidpp20.get_advanced_eq_params(device, direction=0, slot=active_slot)
if not bands:
logger.info("HeadsetAdvancedEQ.build: getCustomEQ returned no bands, no panel will be built")
return None
if len(bands) != band_count:
band_count = len(bands)
# V0/V1 advertises band_count in getEQInfos — cross-check if we have it.
expected = info.get("band_count")
if expected is not None and expected != band_count:
logger.info(
"HeadsetAdvancedEQ.build: band count mismatch — EQInfos=%d getCustomEQ=%d; skipping",
"HeadsetAdvancedEQ.build: V%d band count mismatch — EQInfos=%d getCustomEQ=%d; " "trusting getCustomEQ",
version,
expected,
band_count,
len(bands),
)
return None
keys = common.NamedInts()
for i, (freq, _gain) in enumerate(bands):
keys[i] = str(freq) + _("Hz")
v = cls(keys, min_value=db_min, max_value=db_max, count=band_count, byte_count=1)
v._band_freqs = [freq for freq, _g in bands]
for i, band in enumerate(bands):
freq = band[0]
if version >= 2:
# V2 freq is an opaque u16 bin index — Hz mapping unconfirmed.
keys[i] = _("Band ") + str(i + 1)
else:
keys[i] = str(freq) + _("Hz")
v = cls(
keys,
min_value=int(round(gain_min)),
max_value=int(round(gain_max)),
count=band_count,
byte_count=1,
)
v._version = version
v._step_db = step_db
v._band_freqs = [band[0] for band in bands]
v._band_qs = [band[2] if len(band) >= 3 else 0 for band in bands]
v._active_slot = active_slot
logger.info(
"HeadsetAdvancedEQ.build: panel built with %d band(s), slot=%d, range=[%d,%d]",
"HeadsetAdvancedEQ.build: panel built V%d with %d band(s), slot=%d, range=[%d,%d], step_db=%.4f",
version,
band_count,
active_slot,
db_min,
db_max,
gain_min,
gain_max,
step_db,
)
return v
def validate_read(self, reply_bytes):
# Response: tight-packed 3-byte entries [freq_hi, freq_lo, gain].
if reply_bytes is None:
return {}
version = getattr(self, "_version", 0)
step_db = getattr(self, "_step_db", 1.0)
if version >= 2:
bands, _header_len = hidpp20.parse_v2_bands(reply_bytes, step_db)
if bands is None:
return {}
result = {}
for i, (freq, gain_db, q_u16) in enumerate(bands):
if i >= self.count:
break
result[i] = int(round(gain_db))
if hasattr(self, "_band_freqs") and i < len(self._band_freqs):
self._band_freqs[i] = freq
if hasattr(self, "_band_qs") and i < len(self._band_qs):
self._band_qs[i] = q_u16
return result
# V0/V1: 3-byte stride.
result = {}
offset = 0
i = 0