diff --git a/lib/logitech_receiver/advanced_para_eq.py b/lib/logitech_receiver/advanced_para_eq.py index c8c2ab58..3e0783e5 100644 --- a/lib/logitech_receiver/advanced_para_eq.py +++ b/lib/logitech_receiver/advanced_para_eq.py @@ -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 u16→Hz / u16→Q 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", diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index a5aaffb6..71d571e4 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/lib/logitech_receiver/hidpp20.py @@ -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 diff --git a/lib/logitech_receiver/settings_templates.py b/lib/logitech_receiver/settings_templates.py index 4d1ab513..cb257b7f 100644 --- a/lib/logitech_receiver/settings_templates.py +++ b/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 — 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. + 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 = {}