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.

lib/logitech_receiver/advanced_para_eq.py

@@ -16,10 +16,18 @@
 
 """AdvancedParaEQ (0x020D) helpers.
 
-Unlike OnboardEQ (0x0636) which requires host-computed biquad coefficients,
+The device handles biquad coefficient computation — we transmit only
-AdvancedParaEQ is handled entirely by the device — we send frequency + gain
+per-band frequency, gain, and (on V2) Q-factor; the DSP does the rest.
-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.
+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):
+def _get_version(device) -> int:
-    """Query HEADSET_ADVANCED_PARA_EQ getEQInfos (function 0).
+    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]
+    caps = result[2]
-    # dbMin / dbMax are signed int8 in the doc.
+    gain_min = struct.unpack("b", bytes([result[3]]))[0]
-    db_min = struct.unpack("b", bytes([result[3]]))[0]
+    gain_max = struct.unpack("b", bytes([result[4]]))[0]
-    db_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,
+        caps,
-        db_min,
+        gain_min,
-        db_max,
+        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):
+def _parse_v2_band_payload(result: bytes):
-    """Query getCustomEQ (function 1) for the given direction and slot.
+    """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),

lib/logitech_receiver/hidpp20.py

@@ -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

lib/logitech_receiver/settings_templates.py

@@ -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
+    Writes are intentionally disabled for now — V2's frequency and Q encodings
-    format matches the protocol doc on real hardware before allowing any
+    are still opaque u16 round-trip values (confirmed via LGHUB RE, see
-    changes that could misconfigure the device DSP. Once confirmed working
+    HEADSET_ADVANCED_PARA_EQ_WIRE_PROTOCOL.md), so we can show the current
-    we'll wire up write() to call setCustomEQ.
+    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.
+    V0/V1: 3-byte band stride [freq_hi, freq_lo, gain_i8], gain is whole dB.
-    Each band entry is [freq_hi, freq_lo, gain_db] (3-byte stride). Unlike
+    V2:    5-byte band stride [freq_hi, freq_lo, gain_i8, q_hi, q_lo], gain
-    OnboardEQ (0x0636), the device handles biquad coefficient computation
+           is `signed_byte × step_db` where step_db comes from getEQInfos.
-    — we just send frequency and gain.
     """
 
     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
+            # Cache so get_advanced_eq_params can look up step_db.
-            # Use the active slot on playback direction so the displayed EQ
+            device._advanced_eq_info = info
-            # matches what the user is actually hearing.
+            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):
+            for i, band in enumerate(bands):
-                keys[i] = str(freq) + _("Hz")
+                freq = band[0]
-            v = cls(keys, min_value=db_min, max_value=db_max, count=band_count, byte_count=1)
+                if version >= 2:
-            v._band_freqs = [freq for freq, _g in bands]
+                    # 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,
+                gain_min,
-                db_max,
+                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