HeadsetAdvancedEQ: enable slot-0 band writes
V2 setCustomEQ wire format is pcap-verified: request payload is [dir, slot, pad=0] + N × [freq_hi, freq_lo, filter, gain_hi, gain_lo]. Gain is offset-binary against the [gain_min, gain_max] / gain_steps grid from getEQInfos. Freq and filter type aren't user-editable today; they're sourced from the build-time cache. Changes: - rw_class.write: injects [dir, slot=active, pad=0] before band bytes and re-queries the active slot each call (matches the read path). - validator.prepare_write: encodes the band payload from the dict the UI hands us, converting int dB → raw u16 against the gain grid. - Setting.write: overrides RangeFieldSetting.write to treat empty-bytes replies as success (HID++ set ops ACK with b""), only None as failure. - Drop the read-only override and stale docstring.
This commit is contained in:
parent
6b61e0ef19
commit
be047fd949
|
|
@ -2322,6 +2322,7 @@ class ForceSensingButtonArray(UserDict):
|
||||||
# --- AdvancedParaEQ (0x020D) — re-exported from advanced_para_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_HP # noqa: E402, F401
|
||||||
from .advanced_para_eq import FILTER_TYPE_PEAKING # noqa: E402, F401
|
from .advanced_para_eq import FILTER_TYPE_PEAKING # noqa: E402, F401
|
||||||
|
from .advanced_para_eq import FILTER_TYPE_PEAKING_G522 # noqa: E402, F401
|
||||||
from .advanced_para_eq import get_advanced_eq_active_slot # 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_defaults # noqa: E402, F401
|
||||||
from .advanced_para_eq import get_advanced_eq_friendly_name # noqa: E402, F401
|
from .advanced_para_eq import get_advanced_eq_friendly_name # noqa: E402, F401
|
||||||
|
|
|
||||||
|
|
@ -1900,38 +1900,45 @@ class HeadsetOnboardEQ(settings.RangeFieldSetting):
|
||||||
|
|
||||||
|
|
||||||
class HeadsetAdvancedEQ(settings.RangeFieldSetting):
|
class HeadsetAdvancedEQ(settings.RangeFieldSetting):
|
||||||
"""Read-only display of the headset's active AdvancedParaEQ (0x020D) bands.
|
"""Per-band gain editor for the headset's active AdvancedParaEQ (0x020D) slot.
|
||||||
|
|
||||||
Writes are intentionally disabled for now. The V2 wire format is known
|
V2 wire format (pcap-verified against G522 LIGHTSPEED):
|
||||||
(see advanced_para_eq.py) but we want a round-trip test on real hardware
|
getCustomEQ response: [dir_echo] + N × [freq_hi, freq_lo, filter, gain_hi, gain_lo]
|
||||||
before enabling user-facing writes that could misconfigure the DSP.
|
setCustomEQ request: [dir, slot, pad=0] + N × [freq_hi, freq_lo, filter, gain_hi, gain_lo]
|
||||||
|
Gain is offset-binary against gain_min..gain_max with `gain_steps`
|
||||||
V0/V1: 3-byte band stride [freq_hi, freq_lo, gain_i8], gain is whole dB.
|
discrete positions (raw=120 ≈ 0 dB on G522's [-6, +6] / 241-step
|
||||||
V2: 5-byte band stride [filter_type, freq_hi, freq_lo, gain_hi, gain_lo],
|
grid). Frequency and filter type are read at build time and not
|
||||||
filter_type 0x00=HP 0x78=peaking, freq is raw Hz, gain is signed
|
user-editable today — UI only exposes per-band gain.
|
||||||
int16 × step_db (step_db from getEQInfos).
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
name = "headset-advanced-eq"
|
name = "headset-advanced-eq"
|
||||||
label = _("Headset Advanced EQ (read-only)")
|
label = _("Headset Advanced EQ")
|
||||||
description = _("Display the headset's active parametric EQ. Writes are disabled pending verification.")
|
description = _("Per-band gain for the headset's active parametric EQ.")
|
||||||
feature = _F.HEADSET_ADVANCED_PARA_EQ
|
feature = _F.HEADSET_ADVANCED_PARA_EQ
|
||||||
rw_options = {"read_fnid": 0x10, "write_fnid": 0x20}
|
rw_options = {"read_fnid": 0x10, "write_fnid": 0x20}
|
||||||
keys_universe = []
|
keys_universe = []
|
||||||
|
|
||||||
class rw_class(settings.FeatureRW):
|
class rw_class(settings.FeatureRW):
|
||||||
"""getCustomEQ takes [direction, slot]; the slot is the *active*
|
"""get/setCustomEQ both take [direction, slot]; on writes the
|
||||||
EQ preset, which the device may have switched while we weren't
|
device additionally expects a single 0x00 padding byte before
|
||||||
looking (G HUB on another machine, an onboard button, etc.).
|
the band payload. The slot is the *active* EQ preset, which the
|
||||||
Re-query it on every read instead of caching slot 0 at build
|
device may have switched while we weren't looking — re-query it
|
||||||
time. Direction is hardcoded to 0 (playback) — mic-side EQ
|
on every read and write instead of caching at build time.
|
||||||
isn't exposed yet."""
|
Direction is hardcoded to 0 (playback); mic-side EQ isn't
|
||||||
|
exposed yet.
|
||||||
|
"""
|
||||||
|
|
||||||
def read(self, device, data_bytes=b""):
|
def read(self, device, data_bytes=b""):
|
||||||
active_slot = hidpp20.get_advanced_eq_active_slot(device, direction=0)
|
active_slot = hidpp20.get_advanced_eq_active_slot(device, direction=0)
|
||||||
self.read_prefix = bytes([0, active_slot if active_slot is not None else 0])
|
self.read_prefix = bytes([0, active_slot if active_slot is not None else 0])
|
||||||
return super().read(device, data_bytes)
|
return super().read(device, data_bytes)
|
||||||
|
|
||||||
|
def write(self, device, data_bytes):
|
||||||
|
active_slot = hidpp20.get_advanced_eq_active_slot(device, direction=0)
|
||||||
|
slot = active_slot if active_slot is not None else 0
|
||||||
|
write_bytes = bytes([0, slot, 0]) + data_bytes
|
||||||
|
return device.feature_request(self.feature, self.write_fnid, write_bytes)
|
||||||
|
|
||||||
class validator_class(settings_validator.PackedRangeValidator):
|
class validator_class(settings_validator.PackedRangeValidator):
|
||||||
kind = settings.Kind.GRAPHIC_EQ
|
kind = settings.Kind.GRAPHIC_EQ
|
||||||
|
|
||||||
|
|
@ -2045,13 +2052,53 @@ class HeadsetAdvancedEQ(settings.RangeFieldSetting):
|
||||||
return result
|
return result
|
||||||
|
|
||||||
def prepare_write(self, new_values):
|
def prepare_write(self, new_values):
|
||||||
# Read-only mode: never actually build a write payload.
|
"""Encode N × [freq_hi, freq_lo, filter, gain_hi, gain_lo].
|
||||||
return None
|
|
||||||
|
new_values is {band_idx: int_gain_dB}. freq and filter type
|
||||||
|
come from the cache captured at build time; gain is mapped
|
||||||
|
from integer dB back to the device's offset-binary raw u16
|
||||||
|
against the [_gain_min, _gain_max] / _gain_steps grid.
|
||||||
|
"""
|
||||||
|
version = getattr(self, "_version", 0)
|
||||||
|
if version < 2:
|
||||||
|
return None
|
||||||
|
gain_min = getattr(self, "_gain_min", -6)
|
||||||
|
gain_max = getattr(self, "_gain_max", 6)
|
||||||
|
steps = getattr(self, "_gain_steps", 241)
|
||||||
|
freqs = getattr(self, "_band_freqs", None) or []
|
||||||
|
types = getattr(self, "_band_types", None) or []
|
||||||
|
if not freqs or not types:
|
||||||
|
return None
|
||||||
|
span = gain_max - gain_min
|
||||||
|
payload = bytearray()
|
||||||
|
for i in range(self.count):
|
||||||
|
freq = freqs[i] if i < len(freqs) else 0
|
||||||
|
filt = types[i] if i < len(types) else hidpp20.FILTER_TYPE_PEAKING_G522
|
||||||
|
gain_db = new_values.get(i, 0)
|
||||||
|
if steps > 1 and span > 0:
|
||||||
|
raw = int(round((gain_db - gain_min) / span * (steps - 1)))
|
||||||
|
else:
|
||||||
|
raw = 0
|
||||||
|
raw = max(0, min(steps - 1, raw))
|
||||||
|
payload += bytes([(freq >> 8) & 0xFF, freq & 0xFF, filt & 0xFF, (raw >> 8) & 0xFF, raw & 0xFF])
|
||||||
|
return bytes(payload)
|
||||||
|
|
||||||
def write(self, map, save=True):
|
def write(self, map, save=True):
|
||||||
# Read-only for now — log attempted writes but don't transmit anything.
|
# RangeFieldSetting.write treats an empty-bytes reply (`not reply`)
|
||||||
logger.info("HeadsetAdvancedEQ: write ignored (read-only mode); requested=%s", map)
|
# as failure, but setCustomEQ returns an empty ACK on success.
|
||||||
return None
|
# Override to treat only `reply is None` (transport error/timeout)
|
||||||
|
# as failure.
|
||||||
|
assert hasattr(self, "_value")
|
||||||
|
assert hasattr(self, "_device")
|
||||||
|
assert map is not None
|
||||||
|
if self._device.online:
|
||||||
|
self.update(map, save)
|
||||||
|
data_bytes = self._validator.prepare_write(self._value)
|
||||||
|
if data_bytes is not None:
|
||||||
|
reply = self._rw.write(self._device, data_bytes)
|
||||||
|
if reply is None:
|
||||||
|
return None
|
||||||
|
return map
|
||||||
|
|
||||||
|
|
||||||
class HeadsetActiveEQPreset(settings.Setting):
|
class HeadsetActiveEQPreset(settings.Setting):
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue