diff --git a/CHANGELOG.md b/CHANGELOG.md index 90b5be2..dd59507 100644 --- a/CHANGELOG.md +++ b/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, diff --git a/README.md b/README.md index dea69b3..fbd8de8 100644 --- a/README.md +++ b/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 | diff --git a/docs/protocols.md b/docs/protocols.md index 34caf99..c33b734 100644 --- a/docs/protocols.md +++ b/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) diff --git a/halo_battery.pyw b/halo_battery.pyw index 302d21d..9bcaddb 100644 --- a/halo_battery.pyw +++ b/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 diff --git a/providers/__init__.py b/providers/__init__.py index b28ed07..0557238 100644 --- a/providers/__init__.py +++ b/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 diff --git a/providers/logitech_centurion.py b/providers/logitech_centurion.py new file mode 100644 index 0000000..29bf369 --- /dev/null +++ b/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 = payload length + 1; the payload of a + received frame is frame[3 : 2 + len]. + * a direct request to the receiver: payload , + 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 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 + the id of feature n (data[1:3]). The CenturionBridge (0x0003) is how the + headset is reached. + * a bridge request: payload 0x11 , + sub-message 00 . 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) diff --git a/tests/test_logitech_centurion.py b/tests/test_logitech_centurion.py new file mode 100644 index 0000000..5005263 --- /dev/null +++ b/tests/test_logitech_centurion.py @@ -0,0 +1,304 @@ +"""Tests for providers/logitech_centurion.py (Logitech G PRO X 2 LIGHTSPEED). No hardware. + +The fake receiver speaks the Centurion protocol the way Solaar (base.py, device.py) and +HeadsetControl (logitech_centurion_protocol.hpp) expect it: direct requests to the +receiver's own features, and bridge requests that the receiver acknowledges and the +headset answers. Its feature layout is the one in Solaar's dump of a real PRO X 2 +(docs/devices/PRO X 2 LIGHTSPEED 0AF7.text): bridge at receiver index 3, battery at +headset index 4. +A fake clock replaces time. + +Run from the repository root: + + python -m unittest discover -s tests +""" +import os +import sys +import types +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from providers import logitech_centurion as L # noqa: E402 + + +class Clock: + def __init__(self): + self.now = 1000.0 + + def time(self): + return self.now + + def sleep(self, s): + self.now += s + + +def out_frame(payload): + return L.frame(list(payload)) + + +class Receiver: + """mode: "on", "off" (acks only), "legacy" (no 0x0104 in the headset).""" + + RX = [0x0000, 0x0001, 0x0100, 0x0003, 0x010A] + + def __init__(self, clock, level=80, state=0, mode="on", noise=()): + self.clock = clock + self.level, self.state, self.mode = level, state, mode + self.hs = [0x0000, 0x0001, 0x0101, 0x0100, 0x0104, 0x010A, 0x0108, 0x0604, 0x0602, 0x0636] + if mode == "legacy": + self.hs[4] = 0x0105 # no battery feature + self.queue = [list(r) for r in noise] + self.writes = [] + + def write(self, data): + data = list(data) + self.writes.append(data) + if data[:10] == L.LEGACY_REQUEST: + self.queue += [[0x51, 0x03] + [0] * 62, + [0x51, 0x0B, 0, 0, 0, 0, 0, 0, 0x04, 0, self.level, 0, + 0x02 if self.state else 0] + [0] * 51] + return len(data) + p = L.payload_of(data) + if p[0] == 3 and p[1] == 0x11: # bridge + sub = p[4:4 + (p[2] << 8 | p[3])] + self.queue.append(out_frame([3, 0x11])) # the receiver's ack + if self.mode == "off": + return len(data) + idx, fn = sub[1], sub[2] & 0xF0 + d = self.answer(self.hs, idx, fn, sub[3:]) + if d is None: + self.queue.append(out_frame([3, 0x10, 0, 4, 0x00, 0xFF, idx])) + else: + self.queue.append(out_frame([3, 0x10, 0, 3 + len(d), 0x00, idx, sub[2]] + d)) + else: # direct + d = self.answer(self.RX, p[0], p[1] & 0xF0, p[2:]) + self.queue.append(out_frame([p[0], p[1]] + (d or []))) + return len(data) + + def answer(self, table, idx, fn, params): + if idx == 0 and fn == 0x00: # Root: where is feature X? + want = params[0] << 8 | params[1] + return [table.index(want)] if want in table else [0] + if idx == 1 and fn == 0x00: + return [len(table)] + if idx == 1 and fn == 0x10: + f = table[params[0]] + return [params[0], f >> 8, f & 0xFF, 0] + if idx < len(table) and table[idx] == 0x0104 and fn == 0x00: + return [self.level, 0, self.state] + return None + + def read(self, timeout_ms): + if self.queue: + return self.queue.pop(0) + self.clock.now += timeout_ms / 1000 + return [] + + +class FakeBus: + def __init__(self, devs): + self.devs = devs + + def device_class(self): + bus = self + + class FakeDevice: + def open_path(self, path): + self.d = bus.devs[path] + + def write(self, data): + return self.d.write(data) + + def read(self, n, timeout_ms=0): + return self.d.read(timeout_ms) + + def close(self): + pass + + return FakeDevice + + +def issue_103_entries(pid=0x0AF7): + return [{"product_id": pid, "interface_number": 3, "usage_page": p, "usage": 1, + "path": b"%04x" % p, "product_string": "PRO X 2 LIGHTSPEED"} + for p in (0xFF13, 0xFFA0, 0x000C)] + + +class CenturionTest(unittest.TestCase): + def setUp(self): + self._saved = (L.hid, L.hidlist, L.time) + self.clock = Clock() + L.time = types.SimpleNamespace(time=self.clock.time, sleep=self.clock.sleep) + self.provider = L.LogitechCenturionProvider() + + def tearDown(self): + L.hid, L.hidlist, L.time = self._saved + + def poll(self, rx, entries=None): + entries = issue_103_entries() if entries is None else entries + L.hid = types.SimpleNamespace(device=FakeBus({b"ffa0": rx}).device_class()) + L.hidlist = types.SimpleNamespace(enumerate=lambda vid=0: list(entries)) + return self.provider.poll() + + def test_issue_103(self): + rx = Receiver(self.clock, level=80) + res = self.poll(rx) + self.assertEqual([(r.name, r.level, r.charging, r.online, r.kind) for r in res], + [("Logitech G PRO X 2 LIGHTSPEED", 80, False, True, "headset")]) + + def test_frames_are_headsetcontrols(self): + # direct, as Solaar frames it: len counts the unpadded request + the flags byte + self.assertEqual(L.direct_frame(0, 0x00, [0x00, 0x01])[:8], + [0x51, 0x05, 0x00, 0x00, 0x01, 0x00, 0x01, 0x00]) + self.assertEqual(set(L.direct_frame(0, 0x00, [0x00, 0x01])[7:]), {0}) + self.assertEqual(len(L.direct_frame(0, 0x00)), 64) + # bridge to headset feature 4 (the battery): HeadsetControl's fixed legacy frame is + # this frame with software id 0x0A instead of 0x01 in both function bytes + b = L.bridge_frame(3, 4, 0x00) + self.assertEqual(b[:10], [0x51, 0x08, 0x00, 0x03, 0x11, 0x00, 0x03, 0x00, 0x04, 0x01]) + swapped = [x if i not in (4, 9) else (x & 0xF0) | 0x0A for i, x in enumerate(b[:10])] + self.assertEqual(swapped, L.LEGACY_REQUEST) + + def test_only_reading_functions_are_sent(self): + rx = Receiver(self.clock) + self.poll(rx) + self.poll(rx) + for w in rx.writes: + p = L.payload_of(w) + if p[0] == 3 and p[1] == 0x11: + sub_index, fn = p[5], p[6] & 0xF0 + feature = rx.hs[sub_index] + self.assertIn((feature, fn), {(0x0000, 0x00), (0x0001, 0x00), (0x0001, 0x10), + (0x0104, 0x00)}) + else: + self.assertIn((rx.RX[p[0]], p[1] & 0xF0), {(0x0000, 0x00), (0x0001, 0x00), + (0x0001, 0x10)}) + + def test_discovery_runs_once(self): + rx = Receiver(self.clock) + self.poll(rx) + n = len(rx.writes) + self.poll(rx) + self.assertEqual(len(rx.writes) - n, 1) # only the battery request + + def test_charging_states(self): + for state, flag in ((0, False), (1, True), (2, True), (3, True), (4, False)): + self.provider = L.LogitechCenturionProvider() + res = self.poll(Receiver(self.clock, state=state)) + self.assertEqual([r.charging for r in res], [flag], state) + + def test_headset_off_gives_no_icon_and_is_discovered_later(self): + rx = Receiver(self.clock, mode="off") + self.assertEqual(self.poll(rx), []) + self.assertIn("did not answer", " ".join(self.provider.diagnostics())) + rx.mode = "on" + self.assertEqual([r.level for r in self.poll(rx)], [80]) + + def test_headset_switched_off_after_a_reading_leaves_the_tray(self): + rx = Receiver(self.clock) + self.poll(rx) + rx.mode = "off" + self.assertEqual(self.poll(rx), []) + + def test_level_above_100_is_refused(self): + self.assertEqual(self.poll(Receiver(self.clock, level=0xC8)), []) + + def test_firmware_without_the_battery_feature_uses_the_fixed_request(self): + rx = Receiver(self.clock, level=55, state=1, mode="legacy") + res = self.poll(rx) + self.assertEqual([(r.level, r.charging) for r in res], [(55, True)]) + self.assertEqual(rx.writes[-1][:10], L.LEGACY_REQUEST) + + def test_legacy_power_off_report(self): + rx = Receiver(self.clock, mode="legacy") + rx.write = lambda data, orig=rx.write: ( + rx.queue.append([0x51, 0x05, 0, 0, 0, 0, 0x00] + [0] * 57) + if list(data[:10]) == L.LEGACY_REQUEST else orig(data)) + self.assertEqual(self.poll(rx), []) + self.assertIn("headset off", " ".join(self.provider.diagnostics())) + + def test_legacy_skips_other_events(self): + rx = Receiver(self.clock, level=55, mode="legacy") + orig = rx.write + + def write(data): + if list(data[:10]) == L.LEGACY_REQUEST: + rx.queue.append([0x51, 0x0B, 0, 0, 0, 0, 0, 0, 0x07, 0, 9, 0, 0] + [0] * 51) + return orig(data) + rx.write = write + self.assertEqual([r.level for r in self.poll(rx)], [55]) + + def test_headset_notifications_and_stale_replies_are_not_the_answer(self): + rx = Receiver(self.clock, level=77) + orig = rx.write + + def write(data): + n = orig(data) + p = L.payload_of(list(data)) + if p[0] == 3 and p[1] == 0x11 and p[5] == 4: # the battery request + rx.queue[1:1] = [ + out_frame([3, 0x10, 0, 4, 0xFF, 4, 0x01, 5]), # a notification + out_frame([3, 0x10, 0, 4, 0x00, 4, 0x11, 6]), # another function + ] + return n + rx.write = write + self.assertEqual([r.level for r in self.poll(rx)], [77]) + + def test_features_are_discovered_again_after_an_error(self): + rx = Receiver(self.clock) + self.poll(rx) + rx.hs[4] = 0x0105 # e.g. new firmware: index 4 moved + orig_answer = rx.answer + rx.answer = lambda table, idx, fn, params: ( + None if table is rx.hs and idx == 4 else orig_answer(table, idx, fn, params)) + self.assertEqual(self.poll(rx), []) # rejected: forget the table + rx.answer = orig_answer + rx.hs[4] = 0x0104 + n = len(rx.writes) + self.assertEqual([r.level for r in self.poll(rx)], [80]) + self.assertGreater(len(rx.writes) - n, 1) # discovery ran again + + def test_other_frames_are_skipped(self): + noise = [[0x11, 0xFF, 0x00] + [0] * 61, out_frame([0x07, 0x01, 0x05])] + rx = Receiver(self.clock, level=64, noise=noise) + self.assertEqual([r.level for r in self.poll(rx)], [64]) + + def test_a_reply_for_another_function_is_not_taken(self): + # a late reply to FeatureSet "count" (fn 0) must not answer "feature n" (fn 0x10) + stale = out_frame([1, 0x01, 9]) + rx = Receiver(self.clock, level=77) + orig = rx.write + + def write(data): + n = orig(data) + p = L.payload_of(list(data)) + if p[0] == 1 and p[1] == 0x11: + rx.queue.insert(0, stale) + return n + rx.write = write + self.assertEqual([r.level for r in self.poll(rx)], [77]) + + def test_a_receiver_error_reply_ends_the_attempt(self): + rx = Receiver(self.clock) + orig = rx.write + + def write(data): + p = L.payload_of(list(data)) + if p[0] == 0: + rx.queue.append(out_frame([0xFF, 0x00, 0x01, 0x05])) + return len(data) + return orig(data) + rx.write = write + self.assertEqual(self.poll(rx), []) + self.assertIn("error", " ".join(self.provider.diagnostics())) + + def test_only_the_ffa0_collection_of_0af7_is_opened(self): + rx = Receiver(self.clock) + entries = issue_103_entries(0x0AFB) + [dict(issue_103_entries()[0])] + self.assertEqual(self.poll(rx, entries), []) + self.assertEqual(rx.writes, []) + + +if __name__ == "__main__": + unittest.main()