HeadsetOnboardEffect: add 0x0621 onboard RGB effect control
RGB headsets (e.g. G522) expose HEADSET_RGB_ONBOARD_EFFECTS (0x0621) — a firmware-played RGB effect on the headset's primary lighting cluster, chosen from six effect types: Fixed, Color Cycle, Color Wave, Breathing, Dual Color, Custom. Add an effect-switching HETERO setting modeled on the keyboard zone effects: a six-way effect picker plus the per-effect parameter widgets (colours, intensity, period, saturation, direction), with fields_map showing only the fields the selected effect uses. The rw_class reads getRGBClusterEffect and writes setRGBClusterEffect; the data class encodes each effect's distinct parameter layout. build() reads getRGBClusterInfo to learn the cluster's supported-effect set and offers only those in the picker; an unparseable reply falls back to offering all six (the firmware rejects any it doesn't support). intensity is a 0-100 percent; saturation is a raw 0-255 byte, matching the keyboard RGB effects. Like the signature/boot effects this runs autonomously on the firmware and is not gated on host LED control. Cluster 0 only — no multi-cluster headset has been seen and the getInfo multi-cluster descriptor stride is unconfirmed.
This commit is contained in:
Ken Sanislo
Peter F. Patel-Schneider
parent
275ad64be1
commit
936991e0b4
|
|
@ -2725,6 +2725,177 @@ class HeadsetSignaturePassiveEffect(_HeadsetSignatureEffectSetting):
|
|||
effect_id = 2
|
||||
|
||||
|
||||
class _HeadsetOnboardEffect:
|
||||
"""A 0x0621 onboard RGB effect: an effect ID plus the parameters that
|
||||
effect uses. Synthetic form [effectId_u16_BE, 7 param bytes]; to_bytes /
|
||||
from_bytes encode the per-effect parameter layout (the rw_class adds the
|
||||
leading clusterIndex). intensity is a 0-100 percent; saturation is a raw
|
||||
0-255 byte (same as the keyboard RGB effects)."""
|
||||
|
||||
def __init__(self, ID=0, color1=0, color2=0, intensity=0, saturation=0, period=0, speed=0, direction=0):
|
||||
self.ID = int(ID)
|
||||
self.intensity = max(0, min(100, int(intensity)))
|
||||
self.saturation = max(0, min(255, int(saturation)))
|
||||
self.period = max(0, min(0xFFFF, int(period)))
|
||||
self.speed = max(0, min(0xFF, int(speed)))
|
||||
self.direction = max(0, min(3, int(direction)))
|
||||
for k, v in (("color1", color1), ("color2", color2)):
|
||||
setattr(self, k, common.ColorInt(int(v) & 0xFFFFFF))
|
||||
|
||||
@classmethod
|
||||
def from_bytes(cls, data, options=None):
|
||||
if data is None or len(data) < 9:
|
||||
return cls()
|
||||
eid = (data[0] << 8) | data[1]
|
||||
p = data[2:9]
|
||||
kw = {"ID": eid}
|
||||
if eid == 0: # Fixed
|
||||
kw["color1"] = (p[0] << 16) | (p[1] << 8) | p[2]
|
||||
elif eid in (1, 2): # ColorCycle / ColorWave
|
||||
kw["intensity"] = p[0]
|
||||
kw["period"] = (p[1] << 8) | p[2]
|
||||
kw["saturation"] = p[3]
|
||||
if eid == 2:
|
||||
kw["direction"] = p[4]
|
||||
elif eid == 3: # Breathing
|
||||
kw["color1"] = (p[0] << 16) | (p[1] << 8) | p[2]
|
||||
kw["period"] = (p[4] << 8) | p[5]
|
||||
elif eid == 4: # DualColor
|
||||
kw["color1"] = (p[0] << 16) | (p[1] << 8) | p[2]
|
||||
kw["color2"] = (p[3] << 16) | (p[4] << 8) | p[5]
|
||||
kw["speed"] = p[6]
|
||||
return cls(**kw)
|
||||
|
||||
def to_bytes(self, options=None):
|
||||
eid = self.ID
|
||||
p = bytearray(7)
|
||||
c1, c2 = int(self.color1), int(self.color2)
|
||||
if eid == 0: # Fixed: R, G, B
|
||||
p[0], p[1], p[2] = (c1 >> 16) & 0xFF, (c1 >> 8) & 0xFF, c1 & 0xFF
|
||||
elif eid in (1, 2): # ColorCycle / ColorWave
|
||||
p[0] = self.intensity
|
||||
p[1], p[2] = (self.period >> 8) & 0xFF, self.period & 0xFF
|
||||
p[3] = self.saturation
|
||||
if eid == 2:
|
||||
p[4] = self.direction
|
||||
elif eid == 3: # Breathing: R, G, B, CE[6]=0, period u16 BE
|
||||
p[0], p[1], p[2] = (c1 >> 16) & 0xFF, (c1 >> 8) & 0xFF, c1 & 0xFF
|
||||
p[4], p[5] = (self.period >> 8) & 0xFF, self.period & 0xFF
|
||||
elif eid == 4: # DualColor: R1,G1,B1, R2,G2,B2, speed
|
||||
p[0], p[1], p[2] = (c1 >> 16) & 0xFF, (c1 >> 8) & 0xFF, c1 & 0xFF
|
||||
p[3], p[4], p[5] = (c2 >> 16) & 0xFF, (c2 >> 8) & 0xFF, c2 & 0xFF
|
||||
p[6] = self.speed
|
||||
# eid 5 Custom: no inline parameters
|
||||
return bytes([(eid >> 8) & 0xFF, eid & 0xFF]) + bytes(p)
|
||||
|
||||
def __eq__(self, other):
|
||||
return isinstance(other, self.__class__) and self.to_bytes() == other.to_bytes()
|
||||
|
||||
def __str__(self):
|
||||
return yaml.dump(self, width=float("inf")).rstrip("\n")
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, loader, node):
|
||||
return cls(**loader.construct_mapping(node))
|
||||
|
||||
@classmethod
|
||||
def to_yaml(cls, dumper, data):
|
||||
return dumper.represent_mapping("!HeadsetOnboardEffect", data.__dict__, flow_style=True)
|
||||
|
||||
|
||||
yaml.SafeLoader.add_constructor("!HeadsetOnboardEffect", _HeadsetOnboardEffect.from_yaml)
|
||||
yaml.add_representer(_HeadsetOnboardEffect, _HeadsetOnboardEffect.to_yaml)
|
||||
|
||||
|
||||
class HeadsetOnboardEffect(settings.Setting):
|
||||
"""The firmware RGB effect a headset runs autonomously on its primary
|
||||
lighting cluster (HEADSET_RGB_ONBOARD_EFFECTS, 0x0621). Build reads the
|
||||
cluster's supported-effect set so the picker offers only those. Like the
|
||||
signature/boot effects this is firmware-driven and not gated on host LED
|
||||
control. Multi-cluster devices (none seen yet) drive only cluster 0."""
|
||||
|
||||
name = "headset-onboard-effect"
|
||||
label = _("Onboard Effect")
|
||||
description = _("Firmware RGB effect the headset plays on its own.")
|
||||
feature = _F.HEADSET_RGB_ONBOARD_EFFECTS
|
||||
|
||||
_CLUSTER = 0
|
||||
_ALL_EFFECTS = (
|
||||
("Fixed", 0),
|
||||
("Color Cycle", 1),
|
||||
("Color Wave", 2),
|
||||
("Breathing", 3),
|
||||
("Dual Color", 4),
|
||||
("Custom", 5),
|
||||
)
|
||||
_EFFECT_FIELDS = {
|
||||
0: ("color1",),
|
||||
1: ("intensity", "period", "saturation"),
|
||||
2: ("intensity", "period", "saturation", "direction"),
|
||||
3: ("color1", "period"),
|
||||
4: ("color1", "color2", "speed"),
|
||||
5: (),
|
||||
}
|
||||
_DIRECTIONS = common.NamedInts(**{"Horizontal": 0, "Vertical": 1, "Reverse Horizontal": 2, "Reverse Vertical": 3})
|
||||
|
||||
class rw_class:
|
||||
kind = settings.FeatureRW.kind
|
||||
|
||||
def __init__(self, feature, cluster):
|
||||
self.feature = feature
|
||||
self._cluster = cluster
|
||||
|
||||
def read(self, device):
|
||||
reply = device.feature_request(self.feature, 0x20, bytes([self._cluster])) # getRGBClusterEffect
|
||||
if reply is None or len(reply) < 10:
|
||||
return None
|
||||
return reply[1:10] # strip clusterIndex -> [effectId_u16, 7 param bytes]
|
||||
|
||||
def write(self, device, data_bytes):
|
||||
# data_bytes is [effectId_u16, 7 param bytes]; prepend clusterIndex
|
||||
reply = device.feature_request(self.feature, 0x30, bytes([self._cluster]) + bytes(data_bytes))
|
||||
return data_bytes if reply is not None else None
|
||||
|
||||
@classmethod
|
||||
def build(cls, device):
|
||||
try:
|
||||
info = device.feature_request(cls.feature, 0x10, bytes([cls._CLUSTER])) # getRGBClusterInfo
|
||||
except exceptions.FeatureCallError:
|
||||
return None
|
||||
if info is None or len(info) < 1:
|
||||
return None
|
||||
# [count, count x {effectId_u16_BE, caps_u16_BE}] — take effectId of each entry
|
||||
supported = []
|
||||
for i in range(info[0]):
|
||||
off = 1 + i * 4
|
||||
if off + 2 > len(info):
|
||||
break
|
||||
eid = (info[off] << 8) | info[off + 1]
|
||||
if 0 <= eid <= 5 and eid not in supported:
|
||||
supported.append(eid)
|
||||
if not supported:
|
||||
# Unparseable reply — offer all six; the firmware rejects any it
|
||||
# doesn't support. See the 0x0621 fallback note in protocol RE.
|
||||
supported = [0, 1, 2, 3, 4, 5]
|
||||
rw = cls.rw_class(cls.feature, cls._CLUSTER)
|
||||
validator = settings_validator.HeteroValidator(data_class=_HeadsetOnboardEffect, options=None)
|
||||
setting = cls(device, rw, validator)
|
||||
id_choices = common.NamedInts(**{name: eid for name, eid in cls._ALL_EFFECTS if eid in supported})
|
||||
id_field = {"name": "ID", "kind": settings.Kind.CHOICE, "label": None, "choices": id_choices}
|
||||
setting.possible_fields = [
|
||||
id_field,
|
||||
{"name": "color1", "kind": settings.Kind.COLOR, "label": _("Primary")},
|
||||
{"name": "color2", "kind": settings.Kind.COLOR, "label": _("Secondary")},
|
||||
{"name": "intensity", "kind": settings.Kind.RANGE, "label": _("Intensity"), "min": 0, "max": 100},
|
||||
{"name": "saturation", "kind": settings.Kind.RANGE, "label": _("Saturation"), "min": 0, "max": 255},
|
||||
{"name": "period", "kind": settings.Kind.RANGE, "label": _("Period"), "min": 0, "max": 0xFFFF},
|
||||
{"name": "speed", "kind": settings.Kind.RANGE, "label": _("Speed"), "min": 0, "max": 0xFF},
|
||||
{"name": "direction", "kind": settings.Kind.CHOICE, "label": _("Direction"), "choices": cls._DIRECTIONS},
|
||||
]
|
||||
setting.fields_map = {eid: (id_choices[eid], {field: 1 for field in cls._EFFECT_FIELDS[eid]}) for eid in supported}
|
||||
return setting
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------
|
||||
# LogiVoice (0x0900 + 0x0901..0x0907) — read-only presentation pass.
|
||||
#
|
||||
|
|
@ -4297,6 +4468,7 @@ SETTINGS: list[settings.Setting] = [
|
|||
HeadsetSignatureStartupEffect,
|
||||
HeadsetSignatureShutdownEffect,
|
||||
HeadsetSignaturePassiveEffect,
|
||||
HeadsetOnboardEffect,
|
||||
*_LOGIVOICE_SETTINGS,
|
||||
]
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue