diff --git a/lib/logitech_receiver/advanced_para_eq.py b/lib/logitech_receiver/advanced_para_eq.py index 2ac30272..e67c8fe3 100644 --- a/lib/logitech_receiver/advanced_para_eq.py +++ b/lib/logitech_receiver/advanced_para_eq.py @@ -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), diff --git a/lib/logitech_receiver/hidpp20.py b/lib/logitech_receiver/hidpp20.py index 15945c70..2cafef46 100644 --- a/lib/logitech_receiver/hidpp20.py +++ b/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 diff --git a/lib/logitech_receiver/settings_templates.py b/lib/logitech_receiver/settings_templates.py index f2a72ad3..db6cf7bb 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 — 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