AdvancedParaEQ V2: correct stride + real Hz labels

The RE pass against lghub_agent.arm64 plus the G522 live probe resolved
the V2 wire format. Key corrections to the previous implementation:

  1. 5-byte stride is [filter_type, freq_hi, freq_lo, gain_hi, gain_lo].
     The initial RE interpretation of [freq_hi, freq_lo, gain, q_hi, q_lo]
     was wrong — the 0x78 byte is a filter-type sentinel (peaking), not
     the high byte of a frequency.
  2. No header before the bands. G522's default "header" was actually
     band 0: a high-pass filter at 20 Hz (filter_type=0x00, freq=0x0014).
     Total is 10 bands (1 HP + 9 peaking at ISO octaves), not 9.
  3. Frequency is raw Hz as BE u16 — no log/ERB/bin transform. 0x4E20
     is exactly 20000 Hz.
  4. Gain is signed BE int16 (not int8), scaled by step_db from
     getEQInfos. ±120 maps to ±6 dB at 0.05 dB/LSB on the G522.
  5. No Q on the wire — firmware-fixed per filter type.

get_advanced_eq_info is unchanged (13-byte V2 decode was already right).
Parser tuple shape is now (filter_type_byte, freq_hz, gain_db) across
both V0/V1 and V2 paths; V0/V1 synthesises filter_type=peaking so the
shape is uniform. Band labels display real Hz — "HP 20Hz", "50Hz",
"125Hz", ..., "20000Hz" on G522.

Stays read-only. Will enable write once we round-trip-test with known
raw bytes.
This commit is contained in:
Ken Sanislo 2026-04-19 12:35:55 -07:00
parent 5934fa1940
commit 169caec941
3 changed files with 108 additions and 101 deletions

View File

@ -17,17 +17,23 @@
"""AdvancedParaEQ (0x020D) helpers.
The device handles biquad coefficient computation we transmit only
per-band frequency, gain, and (on V2) Q-factor; the DSP does the rest.
per-band filter-type + frequency + gain; 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].
V0/V1 wire format: 3-byte band stride [freq_hi, freq_lo, gain_i8],
gain is whole dB; 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 u16Hz / u16Q mappings are
unconfirmed and need a LGHUB pcap to pin down. See
HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md.
V2 wire format: 5-byte band stride [filter_type, freq_hi, freq_lo,
gain_hi, gain_lo] with NO header. Filter types are 0x00=HP (cutoff),
0x78=peaking. Frequency is raw BE u16 in Hz. Gain is signed BE int16
× step_db (step_db from getEQInfos). No Q on the wire firmware-fixed
per filter type. getEQInfos returns 13 bytes with gain bounds + step
count, format enum, XY-support flag, and onboard preset counts.
Authoritative source: HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md (the
V2 layout was confirmed via live G522 probe the default EQ is one
HP filter at 20 Hz plus nine peaking filters at ISO centers 50, 125,
250, 500, 1000, 2500, 5000, 10000, 20000 Hz with all gains at zero).
"""
from __future__ import annotations
@ -42,6 +48,14 @@ logger = logging.getLogger(__name__)
DIRECTION_PLAYBACK = 0
DIRECTION_CAPTURE = 1
# V2 filter-type taxonomy (byte [+0] of each band).
FILTER_TYPE_HP = 0x00
FILTER_TYPE_PEAKING = 0x78
FILTER_TYPE_NAMES = {
FILTER_TYPE_HP: "HP",
FILTER_TYPE_PEAKING: "peaking",
}
def _get_version(device) -> int:
return device.features.get_feature_version(SupportedFeature.HEADSET_ADVANCED_PARA_EQ) or 0
@ -156,50 +170,41 @@ def get_advanced_eq_active_slot(device, direction=DIRECTION_PLAYBACK):
return result[0]
def _parse_v2_band_payload(result: bytes):
"""Locate and parse the 5-byte band stride inside a V2 getCustomEQ response.
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).
"""Parse a V2 getCustomEQ/getEQDefaults response.
Trailing all-zero bands (terminators) are stripped.
Returns list of (filter_type_byte, freq_hz, gain_db) tuples, or None.
Response is N × 5 bytes with no header. Each band is
[filter_type, freq_hi, freq_lo, gain_hi, gain_lo].
"""
payload, header_len = _parse_v2_band_payload(result)
if payload is None:
return None, None
if result is None or len(result) == 0:
return None
if len(result) % 5 != 0:
return 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
for i in range(len(result) // 5):
e = result[i * 5 : (i + 1) * 5]
filter_type = e[0]
freq_hz = (e[1] << 8) | e[2]
gain_int16 = struct.unpack(">h", bytes(e[3:5]))[0]
gain_db = gain_int16 * step_db
bands.append((filter_type, freq_hz, gain_db))
return bands
def _band_label(filter_type_byte: int, freq_hz: int) -> str:
kind = FILTER_TYPE_NAMES.get(filter_type_byte, f"type-0x{filter_type_byte:02X}")
if filter_type_byte == FILTER_TYPE_HP:
return f"HP {freq_hz} Hz"
return f"{freq_hz} Hz" if kind == "peaking" else f"{kind} {freq_hz} Hz"
def get_advanced_eq_defaults(device, direction=DIRECTION_PLAYBACK, slot=0):
"""Query getEQDefaults (function 5). Same per-band layout as getCustomEQ.
Factory presets, read-only. Reading them across all factory slots gives
us a corpus of (name, freq_u16[], q_u16[]) tuples that may reveal the
u16->Hz and u16->Q scaling without needing a pcap. Returns list of
(freq_u16, gain_db, q_u16) or None.
Returns list of (filter_type_byte, freq_hz, gain_db) tuples, or None.
V0/V1 callers receive (FILTER_TYPE_PEAKING, freq_hz, gain_db) for
compatibility with the V2 tuple shape.
"""
version = _get_version(device)
result = device.feature_request(SupportedFeature.HEADSET_ADVANCED_PARA_EQ, 0x50, direction, slot)
@ -214,25 +219,25 @@ def get_advanced_eq_defaults(device, direction=DIRECTION_PLAYBACK, slot=0):
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
bands, header_len = parse_v2_bands(result, step_db)
bands = parse_v2_bands(result, step_db)
if bands is None:
logger.info(
"AdvancedParaEQ getEQDefaults V2 (dir=%d slot=%d): couldn't locate band payload raw=%s",
"AdvancedParaEQ getEQDefaults V2 (dir=%d slot=%d): payload not multiple of 5 raw=%s",
direction,
slot,
result.hex(),
)
return None
logger.info(
"AdvancedParaEQ getEQDefaults V2 (dir=%d slot=%d): %d band(s) header_len=%d raw=%s",
"AdvancedParaEQ getEQDefaults V2 (dir=%d slot=%d): %d band(s) %s raw=%s",
direction,
slot,
len(bands),
header_len,
[_band_label(t, f) + f" {round(g, 2)}dB" for t, f, g in bands],
result.hex(),
)
return bands
# V0/V1: 3-byte stride, gain is whole dB, no Q.
# V0/V1 legacy 3-byte stride.
bands = []
offset = 0
while offset + 3 <= len(result):
@ -240,7 +245,7 @@ def get_advanced_eq_defaults(device, direction=DIRECTION_PLAYBACK, slot=0):
if freq == 0:
break
gain_db = struct.unpack("b", bytes([result[offset + 2]]))[0]
bands.append((freq, float(gain_db), 0))
bands.append((FILTER_TYPE_PEAKING, freq, float(gain_db)))
offset += 3
logger.info(
"AdvancedParaEQ getEQDefaults V%d (dir=%d slot=%d): %d band(s) raw=%s",
@ -269,19 +274,17 @@ def get_advanced_eq_friendly_name(device, direction=DIRECTION_PLAYBACK, slot=0):
def probe_all_presets(device, direction=DIRECTION_PLAYBACK):
"""Read every factory and custom preset slot's name + band data and log it.
"""Read every factory + custom preset slot and log name + band data at INFO.
Diagnostic probe intended to run once at HeadsetAdvancedEQ.build() time
so we accumulate a corpus of (name, freq_u16[], q_u16[]) tuples across
presets. Patterns in that corpus should reveal the u16->Hz and u16->Q
mappings without needing a live pcap from LGHUB.
Diagnostic probe intended to run once at HeadsetAdvancedEQ.build() time.
The logged corpus is useful for spotting filter-type or frequency pattern
differences between named presets.
"""
info = getattr(device, "_advanced_eq_info", None)
if not info:
return
ro_count = info.get("onboard_ro_preset_count", 0)
custom_count = info.get("onboard_custom_preset_count", 0)
# Factory presets: read via getEQDefaults
for slot in range(ro_count):
name = get_advanced_eq_friendly_name(device, direction=direction, slot=slot)
bands = get_advanced_eq_defaults(device, direction=direction, slot=slot)
@ -293,10 +296,8 @@ def probe_all_presets(device, direction=DIRECTION_PLAYBACK):
ro_count,
direction,
name,
[(f"0x{f:04X}", round(g, 3), f"0x{q:04X}") for f, g, q in bands],
[f"{_band_label(t, f)} {round(g, 2)}dB" for t, f, g in bands],
)
# Custom preset slots: via getCustomEQ. Most will be empty/default, but
# any user-authored slots could give additional freq/Q samples.
for slot in range(custom_count):
name = get_advanced_eq_friendly_name(device, direction=direction, slot=slot)
bands = get_advanced_eq_params(device, direction=direction, slot=slot)
@ -308,25 +309,29 @@ def probe_all_presets(device, direction=DIRECTION_PLAYBACK):
custom_count,
direction,
name,
[(f"0x{f:04X}", round(g, 3), f"0x{q:04X}") for f, g, q in bands],
[f"{_band_label(t, f)} {round(g, 2)}dB" for t, f, g in bands],
)
def get_advanced_eq_params(device, direction=DIRECTION_PLAYBACK, slot=0):
"""Query getCustomEQ (function 1). Returns list of (freq, gain_db, q) or None.
"""Query getCustomEQ (function 1). Returns list of (filter_type, freq_hz, gain_db) 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.
V0/V1: filter_type is always FILTER_TYPE_PEAKING (synthesized), freq is
raw Hz from wire, gain is whole dB.
V2: filter_type comes from the wire (0x00=HP, 0x78=peaking), freq is raw
Hz, gain is int16 × step_db.
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.
step_db for V2 is cached on the device by get_advanced_eq_info.
"""
version = _get_version(device)
result = device.feature_request(SupportedFeature.HEADSET_ADVANCED_PARA_EQ, 0x10, direction, slot)
if result is None:
logger.info("AdvancedParaEQ getCustomEQ V%d (dir=%d slot=%d): feature_request returned None", version, direction, slot)
logger.info(
"AdvancedParaEQ getCustomEQ V%d (dir=%d slot=%d): feature_request returned None",
version,
direction,
slot,
)
return None
if version >= 2:
@ -336,22 +341,27 @@ def get_advanced_eq_params(device, direction=DIRECTION_PLAYBACK, slot=0):
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)
bands = parse_v2_bands(result, step_db)
if bands is None:
logger.info("AdvancedParaEQ getCustomEQ V2: couldn't locate band payload raw=%s", result.hex())
logger.info(
"AdvancedParaEQ getCustomEQ V2 (dir=%d slot=%d): payload not multiple of 5 raw=%s",
direction,
slot,
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",
"AdvancedParaEQ getCustomEQ V2 (dir=%d slot=%d): %d band(s) step_db=%.4f %s raw=%s",
direction,
slot,
len(bands),
header_len,
step_db,
[f"{_band_label(t, f)} {round(g, 2)}dB" for t, f, g in bands],
result.hex(),
)
return bands
# V0 / V1: 3-byte stride, freq is raw Hz, gain is whole dB, no Q.
# V0 / V1
bands = []
offset = 0
while offset + 3 <= len(result):
@ -359,7 +369,7 @@ def get_advanced_eq_params(device, direction=DIRECTION_PLAYBACK, slot=0):
if freq == 0:
break
gain_db = struct.unpack("b", bytes([result[offset + 2]]))[0]
bands.append((freq, float(gain_db), 0))
bands.append((FILTER_TYPE_PEAKING, freq, float(gain_db)))
offset += 3
logger.info(
"AdvancedParaEQ getCustomEQ V%d (dir=%d slot=%d): parsed %d band(s) %s raw=%s",

View File

@ -2304,6 +2304,8 @@ class ForceSensingButtonArray(UserDict):
# --- OnboardEQ (0x0636) — re-exported from onboard_eq.py ---
# --- AdvancedParaEQ (0x020D) — re-exported from advanced_para_eq.py ---
from .advanced_para_eq import FILTER_TYPE_HP # noqa: E402, F401
from .advanced_para_eq import FILTER_TYPE_PEAKING # noqa: E402, F401
from .advanced_para_eq import get_advanced_eq_active_slot # noqa: E402, F401
from .advanced_para_eq import get_advanced_eq_defaults # noqa: E402, F401
from .advanced_para_eq import get_advanced_eq_friendly_name # noqa: E402, F401

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 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 u16Hz and u16Q mappings.
Writes are intentionally disabled for now. We now know the V2 wire format
(see HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md) so a write path is
buildable, but we still want a round-trip test on real hardware before
enabling user-facing writes that could misconfigure the DSP.
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.
V2: 5-byte band stride [filter_type, freq_hi, freq_lo, gain_hi, gain_lo],
filter_type 0x00=HP 0x78=peaking, freq is raw Hz, gain is signed
int16 × step_db (step_db from getEQInfos).
"""
name = "headset-advanced-eq"
@ -1882,7 +1882,6 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
if not info:
logger.info("HeadsetAdvancedEQ.build: getEQInfos failed, no panel will be built")
return None
# 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"]
@ -1895,24 +1894,21 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
logger.info("HeadsetAdvancedEQ.build: getCustomEQ returned no bands, no panel will be built")
return None
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: V%d band count mismatch — EQInfos=%d getCustomEQ=%d; " "trusting getCustomEQ",
"HeadsetAdvancedEQ.build: V%d band count mismatch — EQInfos=%d getCustomEQ=%d; trusting getCustomEQ",
version,
expected,
band_count,
)
keys = common.NamedInts()
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)
for i, (filter_type, freq_hz, _gain_db) in enumerate(bands):
if filter_type == hidpp20.FILTER_TYPE_HP:
keys[i] = "HP " + str(freq_hz) + _("Hz")
else:
keys[i] = str(freq) + _("Hz")
keys[i] = str(freq_hz) + _("Hz")
v = cls(
keys,
min_value=int(round(gain_min)),
@ -1922,8 +1918,8 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
)
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._band_types = [band[0] for band in bands]
v._band_freqs = [band[1] for band in bands]
v._active_slot = active_slot
logger.info(
"HeadsetAdvancedEQ.build: panel built V%d with %d band(s), slot=%d, range=[%d,%d], step_db=%.4f",
@ -1934,10 +1930,9 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
gain_max,
step_db,
)
# One-shot corpus probe: read every factory/custom preset's name
# and band data. Comparing freq_u16 and q_u16 values across named
# presets ("Flat" vs "Bass Boost" etc.) may reveal the u16->Hz
# and u16->Q scalings without requiring a LGHUB pcap.
# One-shot corpus probe — logs every factory + custom preset's
# band data at INFO. Useful diagnostic if a device turns up with
# filter types beyond the observed 0x00/0x78 pair.
if version >= 2:
try:
hidpp20.probe_advanced_eq_presets(device, direction=0)
@ -1951,18 +1946,18 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
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)
bands = hidpp20.parse_v2_bands(reply_bytes, step_db)
if bands is None:
return {}
result = {}
for i, (freq, gain_db, q_u16) in enumerate(bands):
for i, (filter_type, freq_hz, gain_db) in enumerate(bands):
if i >= self.count:
break
result[i] = int(round(gain_db))
if hasattr(self, "_band_types") and i < len(self._band_types):
self._band_types[i] = filter_type
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
self._band_freqs[i] = freq_hz
return result
# V0/V1: 3-byte stride.
result = {}