Logitech G PRO X 2 LIGHTSPEED headset over its receiver (Centurion, #103), unit tests

CHANGELOG.md

@@ -6,6 +6,11 @@ and the project follows [Semantic Versioning](https://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- Logitech G PRO X 2 LIGHTSPEED headset on its receiver (046D:0AF7, #103), over Logitech's
+  Centurion transport as Solaar and HeadsetControl read it: battery and charging, read-only
+  requests. Confirmed on a real headset (#103).
+
 ## [1.13.0] - 2026-09-29
 
 A new Windows 11 style tray menu and a batch of tray features: turn device types off,

README.md

@@ -36,6 +36,7 @@ Levels are read over USB/HID (a dongle, a receiver or a cable), from Xbox-style
 | [Keychron Ultra-Link 8K, Keychron M5](docs/protocols.md#keychron-ultra-link-8k-keychron-m5) | 2.4 GHz receiver and USB cable | no |
 | [LAMZU Maya X](docs/protocols.md#lamzu-maya-x) | 8K dongle or USB cable | yes |
 | [Lofree Hyzen](docs/protocols.md#lofree-hyzen) | 2.4 GHz dongle | no |
+| [Logitech G PRO X 2 LIGHTSPEED](docs/protocols.md#logitech-g-pro-x-2-lightspeed) | 2.4 GHz receiver | yes |
 | [Logitech G502 LIGHTSPEED, G502 X PLUS](docs/protocols.md#logitech-g502-lightspeed-g502-x-plus) | Lightspeed receiver | yes |
 | [Logitech (more HID++ 2.0 devices and G-series headsets)](docs/protocols.md#logitech-more-hid-20-devices-and-g-series-headsets) | Lightspeed, Unifying or Bolt receiver | likely |
 | [MCHOSE A7 V2 Ultra](docs/protocols.md#mchose-a7-v2-ultra) | 2.4 GHz receiver | no |

docs/protocols.md

@@ -48,6 +48,12 @@ A different protocol from the Cloud II: the vendor collection is `0xFF13:0x0001`
 
 The receiver exposes one vendor collection, `ff13:0001` (interface 5 on a real unit), and the headset pushes its own reports there; there is no request to send. The battery arrives as report 0x08 with the level in byte 1 - the pattern plugato/JBL_Baterry_Monitor confirmed on this USB id - and report 0x2f is the microphone mute state. The headset can stay quiet for long stretches, so the last level heard is kept and shown greyed out until something arrives. **Confirmed on a real unit**: the provider finds the receiver, matches its collection and shows the level as soon as the headset reports. That report is an event - on the unit tested it arrives when the headset is plugged into its charger (a capture read 95% at the moment JBL's own app said 95%), and pressing the buttons or the volume rocker does not produce it - so the last level heard is kept and shown greyed out until the next one. The receiver also pushes a power report (0x09 - byte 1 is `0x00` when the headset is switched off and `0x01` when it is on, confirmed on a real unit - plus a 0x02 frame); none of them is a level, and a headset last seen switched off says so in the diagnostics rather than only "nothing heard"
 
+### Logitech G PRO X 2 LIGHTSPEED
+
+**Connection:** 2.4 GHz receiver (046D:0AF7)
+
+Not HID++ on ff43: Logitech's "Centurion" transport on the vendor collection `ffa0:0001`, report 0x51. The app lists the receiver's features, reaches the headset through the receiver's bridge feature (0x0003), lists the headset's features and reads its battery feature 0x0104 (percent, charging state), all with read-only functions. From Solaar (tested on this headset) and HeadsetControl; firmware without 0x0104 gets HeadsetControl's fixed request. Confirmed on hardware in #103 (75 %, level and charging equal to G HUB)
+
 ### Razer Barracuda Pro (2.4 GHz)
 
 **Connection:** 2.4 GHz dongle (1532:053a)

halo_battery.pyw

@@ -78,6 +78,7 @@ from providers import (AmInfinityProvider, AstroProvider, AsusProvider,  # noqa:
                        EightBitDoProvider,
                        GWolvesProvider, HyperXAlpha2Provider, HyperXCloud3Provider, HyperXProvider, JblProvider,
                        KeychronProvider, LamzuProvider, LofreeProvider, LogitechProvider,
+                       LogitechCenturionProvider,
                        MchoseProvider, NintendoProvider, PlayStationProvider, PulsarProvider,
                        RazerProvider, SteelSeriesProvider, WLmouseProvider, XInputProvider)
 from providers.bluetooth import BluetoothWatcher  # noqa: E402
@@ -130,6 +131,7 @@ PROVIDER_LABELS = {
     "lamzu": "LAMZU mice",
     "lofree": "Lofree keyboards",
     "logitech": "Logitech",
+    "logitech_centurion": "Logitech G PRO X 2 LIGHTSPEED",
     "mchose": "MCHOSE mice",
     "nintendo": "Nintendo Switch controllers",
     "playstation": "PlayStation controllers",
@@ -148,7 +150,7 @@ def make_providers() -> list:
             JblProvider(), LogitechProvider(), SteelSeriesProvider(), XInputProvider(),
             PlayStationProvider(), EightBitDoProvider(), BarracudaProvider(), NintendoProvider(),
             AsusProvider(), GWolvesProvider(), LofreeProvider(), AstroProvider(), CorsairProvider(),
-            LamzuProvider(), AmInfinityProvider()]
+            LamzuProvider(), AmInfinityProvider(), LogitechCenturionProvider()]
 
 
 # ---------------------------------------------------------------- config

providers/__init__.py

@@ -7,6 +7,7 @@
 from .bluetooth import BluetoothProvider  # noqa: F401
 from .xinput import XInputProvider  # noqa: F401
 from .logitech import LogitechProvider  # noqa: F401
+from .logitech_centurion import LogitechCenturionProvider  # noqa: F401
 from .steelseries import SteelSeriesProvider  # noqa: F401
 from .jbl import JblProvider  # noqa: F401
 from .keychron import KeychronProvider  # noqa: F401

providers/logitech_centurion.py

@@ -0,0 +1,279 @@
+"""Logitech G PRO X 2 LIGHTSPEED headset (046D:0AF7) over its receiver, without G HUB.
+
+This headset does not use HID++ on the receiver's ff43 collection (logitech.py). Sources,
+which agree on everything below except where noted:
+  * pwr-Solaar/Solaar at e7304c4: lib/logitech_receiver/base.py (framing),
+    device.py (centurion_bridge_request), centurion.py (battery), and
+    docs/devices/PRO X 2 LIGHTSPEED 0AF7.text, a dump of a real headset: receiver
+    features 0 Root, 1 FeatureSet, 2 0x0100, 3 CentPP bridge, 4 0x010A; headset
+    feature 4 = 0x0104 battery.
+  * Sapd/HeadsetControl at 25dadae: lib/devices/logitech_gpro_x2_lightspeed.hpp and
+    lib/devices/protocols/logitech_centurion_protocol.hpp ("Centurion").
+
+  * the vendor collection usage page 0xFFA0 / usage 0x0001 (interface 3), 64-byte
+    output and input reports with report id 0x51.
+  * a frame: 51 <len> <flags> <payload...>, len = payload length + 1; the payload of a
+    received frame is frame[3 : 2 + len].
+  * a direct request to the receiver: payload <feature index> <function | sw id> <params>,
+    sw id 1, and len counts only these bytes, as in Solaar (HeadsetControl frames a
+    zero-padded 64-byte buffer, len 0x41). The reply echoes the first two bytes; its
+    data is payload[2:]. An error reply is ff <feature index> <function> <code>.
+  * feature discovery on the receiver: Root (index 0) function 0 with the id 0x0001
+    gives the FeatureSet index; FeatureSet function 0 gives the count and function 0x10
+    <n> the id of feature n (data[1:3]). The CenturionBridge (0x0003) is how the
+    headset is reached.
+  * a bridge request: payload <bridge index> 0x11 <size hi> <size lo> <sub-message>,
+    sub-message 00 <feature index> <function | sw id> <params>. The receiver first
+    acknowledges (bridge index, 0x11); the headset's answer is (bridge index, 0x10),
+    then 2 size bytes, 00, the feature index (0xFF = rejected), the echoed function byte,
+    and the data.
+  * the same discovery through the bridge gives the headset's own features. The battery
+    is CenturionBatterySoc (0x0104), function 0: data[0] = percent, data[2] = charging
+    state: 0 discharging, 1 charging, 2 charging over USB, 3 charge complete (Solaar;
+    HeadsetControl names only 1 and 2). 1-3 mean the headset is on the cable.
+  * firmware without 0x0104: HeadsetControl's fixed request 51 08 00 03 1a 00 03 00 04 0a.
+    Up to 4 replies: 51 03 ... is an acknowledgement, 51 05 with byte 6 = 00 means the
+    headset is off, and the battery reply is 51 0b ... with byte 8 = 04: byte 10 percent,
+    byte 12 = 02 charging.
+
+Every request reads: discovery lists features and the battery request reads the level.
+No function that changes a setting is sent. A level above 100 is refused.
+"""
+from __future__ import annotations
+
+import time
+from typing import Dict, List, Optional, Tuple
+
+try:
+    import hid
+except ImportError:              # pragma: no cover
+    hid = None
+
+from . import hidlist
+from .base import DeviceStatus, Provider, hexdump, log
+
+LOGITECH_VID = 0x046D
+PIDS = {0x0AF7: "Logitech G PRO X 2 LIGHTSPEED"}
+USAGE_PAGE, USAGE = 0xFFA0, 0x0001
+
+REPORT_ID = 0x51
+FRAME_SIZE = 64
+SOFTWARE_ID = 0x01
+BRIDGE_SEND = 0x10               # function of a bridge send and of its message event
+
+FEATURE_ROOT = 0x0000
+FEATURE_SET = 0x0001
+FEATURE_BRIDGE = 0x0003
+FEATURE_BATTERY = 0x0104
+
+LEGACY_REQUEST = [0x51, 0x08, 0x00, 0x03, 0x1A, 0x00, 0x03, 0x00, 0x04, 0x0A]
+LEGACY_READS = 4
+
+WINDOW = 1.5                     # s to wait for one answer (HeadsetControl allows 5 s per read)
+READ_TIMEOUT_MS = 100
+
+
+class Offline(Exception):
+    """The headset is off or out of range: the receiver answers, the headset does not."""
+
+
+def frame(payload: List[int], flags: int = 0) -> List[int]:
+    out = [0] * FRAME_SIZE
+    out[0], out[1], out[2] = REPORT_ID, len(payload) + 1, flags
+    for i, b in enumerate(payload[:FRAME_SIZE - 3]):
+        out[3 + i] = b
+    return out
+
+
+def direct_frame(feature_index: int, function: int, params=()) -> List[int]:
+    return frame([feature_index, (function & 0xF0) | SOFTWARE_ID] + list(params))
+
+
+def bridge_frame(bridge_index: int, sub_index: int, function: int, params=()) -> List[int]:
+    sub = [0x00, sub_index, (function & 0xF0) | SOFTWARE_ID] + list(params)
+    return frame([bridge_index, BRIDGE_SEND | SOFTWARE_ID, (len(sub) >> 8) & 0x0F,
+                  len(sub) & 0xFF] + sub)
+
+
+def payload_of(r) -> Optional[List[int]]:
+    if not r or len(r) < 4 or r[0] != REPORT_ID:
+        return None
+    n = r[1]
+    if n <= 1 or n + 2 > len(r):
+        return None
+    return list(r[3:2 + n])
+
+
+def parse_battery(data) -> Optional[Tuple[int, bool]]:
+    if not data or data[0] > 100:
+        return None
+    return data[0], len(data) >= 3 and data[2] in (1, 2, 3)
+
+
+def parse_legacy(r) -> Optional[Tuple[int, bool]]:
+    if len(r) >= 13 and r[0] == REPORT_ID and r[1] == 0x0B and r[8] == 0x04 and r[10] <= 100:
+        return r[10], r[12] == 0x02
+    return None
+
+
+class LogitechCenturionProvider(Provider):
+    name = "logitech_centurion"
+
+    def __init__(self):
+        self._diag: List[str] = []
+        # path -> (bridge index, {headset feature id: index}) once discovery succeeded
+        self._features: Dict[bytes, Tuple[int, Dict[int, int]]] = {}
+
+    # ---- transport ---------------------------------------------------------
+    def _write(self, dev, data: List[int]) -> None:
+        dev.write(data)
+
+    def _replies(self, dev):
+        """Frames that arrive within the window."""
+        end = time.time() + WINDOW
+        while time.time() < end:
+            r = list(dev.read(FRAME_SIZE, READ_TIMEOUT_MS) or [])
+            if r:
+                yield r
+
+    def _direct(self, dev, feature_index: int, function: int, params=()) -> List[int]:
+        fn_sw = (function & 0xF0) | SOFTWARE_ID
+        self._write(dev, direct_frame(feature_index, function, params))
+        for r in self._replies(dev):
+            p = payload_of(r)
+            if not p or len(p) < 2:
+                continue
+            if p[0] == 0xFF and len(p) >= 3 and p[1] == feature_index and p[2] == fn_sw:
+                raise ValueError(f"receiver feature {feature_index:#04x}: error "
+                                 f"{p[3] if len(p) > 3 else 0:#04x}")
+            if p[0] == feature_index and p[1] == fn_sw:
+                return p[2:]
+        raise TimeoutError(f"no reply from receiver feature {feature_index:#04x}")
+
+    def _bridge(self, dev, bridge: int, sub_index: int, function: int, params=()) -> List[int]:
+        fn_sw = (function & 0xF0) | SOFTWARE_ID
+        self._write(dev, bridge_frame(bridge, sub_index, function, params))
+        acked = False
+        for r in self._replies(dev):
+            p = payload_of(r)
+            if not p or len(p) < 2 or p[0] != bridge or (p[1] >> 4) != (BRIDGE_SEND >> 4):
+                continue
+            if (p[1] & 0x0F) == SOFTWARE_ID:
+                acked = True                  # the receiver took the message
+                continue
+            if (p[1] & 0x0F) != 0 or len(p) < 7 or p[4] != 0x00:
+                continue
+            if p[5] == 0xFF and p[6] == sub_index:
+                raise ValueError(f"headset rejected feature {sub_index:#04x}")
+            if p[5] == sub_index and p[6] == fn_sw:
+                return p[7:]
+        if acked:
+            raise Offline("the receiver acknowledged, the headset did not answer")
+        raise TimeoutError("no acknowledgement from the receiver")
+
+    # ---- discovery -----------------------------------------------------------
+    def _discover(self, dev) -> Tuple[int, Dict[int, int]]:
+        ids = [FEATURE_SET >> 8, FEATURE_SET & 0xFF]
+        fs = self._direct(dev, FEATURE_ROOT, 0x00, ids)
+        if not fs or fs[0] == 0:
+            raise ValueError("receiver: no FeatureSet")
+        count = self._direct(dev, fs[0], 0x00)[0]
+        bridge = None
+        for n in range(count):
+            d = self._direct(dev, fs[0], 0x10, [n])
+            if len(d) >= 3 and (d[1] << 8 | d[2]) == FEATURE_BRIDGE:
+                bridge = n
+                break
+        if bridge is None:
+            raise ValueError("receiver: no Centurion bridge")
+        sfs = self._bridge(dev, bridge, FEATURE_ROOT, 0x00, ids)
+        if not sfs or sfs[0] == 0:
+            raise ValueError("headset: no FeatureSet")
+        sub_count = self._bridge(dev, bridge, sfs[0], 0x00)[0]
+        features: Dict[int, int] = {}
+        for n in range(sub_count):
+            d = self._bridge(dev, bridge, sfs[0], 0x10, [n])
+            if len(d) >= 3:
+                features[d[1] << 8 | d[2]] = n
+        self._diag.append(f"  bridge index {bridge}, headset features: "
+                          + ", ".join(f"{f:04x}@{i}" for f, i in sorted(features.items())))
+        return bridge, features
+
+    def _legacy(self, dev) -> Optional[Tuple[int, bool]]:
+        self._write(dev, LEGACY_REQUEST + [0] * (FRAME_SIZE - len(LEGACY_REQUEST)))
+        seen = 0
+        for r in self._replies(dev):
+            if r[0] != REPORT_ID:
+                continue
+            seen += 1
+            if len(r) >= 7 and r[1] == 0x05 and r[6] == 0x00:
+                raise Offline("power report: headset off")
+            res = parse_legacy(r)
+            if res is not None:
+                return res
+            if seen >= LEGACY_READS:
+                break
+        return None
+
+    def _read(self, path: bytes) -> Optional[Tuple[int, bool]]:
+        dev = hid.device()
+        try:
+            dev.open_path(path)
+        except (OSError, IOError) as e:
+            self._diag.append(f"  open: {e}")
+            return None
+        try:
+            known = self._features.get(path)
+            if known is None:
+                known = self._discover(dev)
+                self._features[path] = known
+            bridge, features = known
+            index = features.get(FEATURE_BATTERY)
+            if index is None:
+                self._diag.append("  no battery feature 0104: the fixed legacy request")
+                return self._legacy(dev)
+            data = self._bridge(dev, bridge, index, 0x00)
+            self._diag.append(f"  battery 0104@{index}: {hexdump(data, 4)}")
+            return parse_battery(data)
+        finally:
+            try:
+                dev.close()
+            except Exception:
+                pass
+
+    # ---- poll ------------------------------------------------------------------
+    def poll(self) -> List[DeviceStatus]:
+        self._diag = []
+        if hid is None:
+            return []
+        try:
+            infos = hidlist.enumerate(LOGITECH_VID)
+        except Exception as e:  # pragma: no cover
+            log.warning("hid.enumerate(logitech centurion): %s", e)
+            return []
+        out: List[DeviceStatus] = []
+        for d in infos:
+            name = PIDS.get(d["product_id"])
+            if name is None or (d.get("usage_page"), d.get("usage")) != (USAGE_PAGE, USAGE):
+                continue
+            key = f"logitech-centurion:{d['product_id']:04x}"
+            self._diag.append(f"[Logitech] pid={d['product_id']:04x} '{name}' "
+                              f"iface={d.get('interface_number')} {USAGE_PAGE:04x}:{USAGE:04x}")
+            try:
+                res = self._read(d["path"])
+            except Offline as e:
+                self._diag.append(f"  {e}")
+                res = None
+            except (TimeoutError, ValueError, OSError, IOError, IndexError) as e:
+                self._diag.append(f"  {e}")
+                self._features.pop(d["path"], None)       # discover again next time
+                res = None
+            # a switched-off headset leaves the tray (README: The icon)
+            if res is not None:
+                level, charging = res
+                out.append(DeviceStatus(key, name, level, charging, True, "logitech",
+                                        kind="headset"))
+        return out
+
+    def diagnostics(self) -> List[str]:
+        return list(self._diag)