diff --git a/CHANGELOG.md b/CHANGELOG.md index efea53e..803f2e8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -118,6 +118,9 @@ and the project follows [Semantic Versioning](https://semver.org/). - HyperX Cloud III Wireless: a dongle that takes the battery request only as a feature report ("Incorrect function" on a normal write) showed no level. The app now notices the refused write and sends the request as a feature report, as intended. +- Razer Barracuda Pro: while the headset was off, each poll waited about 4 seconds longer + than needed and held back the icons of all other devices. The app now stops asking as + soon as the headset does not answer, and it retries when the receiver refuses a command. ## [1.12.0] - 2026-09-28 diff --git a/providers/barracuda.py b/providers/barracuda.py index f9a83ae..0db9817 100644 --- a/providers/barracuda.py +++ b/providers/barracuda.py @@ -131,11 +131,15 @@ def close(self): def _write(self, frame: bytes) -> bool: try: - self.dev.write(frame) - return True + n = self.dev.write(frame) except (OSError, IOError) as e: self.diag.append(f" write: {e}") return False + if n is not None and n < 0: + # hidapi reports a failed write by returning -1, not by raising + self.diag.append(f" write -> {n}") + return False + return True def _drain(self): for _ in range(8): @@ -204,11 +208,13 @@ def read_battery(path: bytes, diag: List[str]) -> Tuple[str, Optional[int], bool return "offline", None, False time.sleep(0.05) battery = s.query(CMD_BATTERY) - charging = s.query(CMD_CHARGING) - s.remote(False) if battery is None: + # no answer: the charging query would only wait out its own timeouts + s.remote(False) diag.append(" no battery reply") return "offline", None, False + charging = s.query(CMD_CHARGING) + s.remote(False) level = parse_level(battery) if level is None: diag.append(f" battery reply out of range: {battery[:4]}") diff --git a/tests/test_barracuda.py b/tests/test_barracuda.py new file mode 100644 index 0000000..8e842f7 --- /dev/null +++ b/tests/test_barracuda.py @@ -0,0 +1,159 @@ +"""Tests for providers/barracuda.py. No hardware is needed. + +The fake receiver records every frame written to it and answers the 'P','A' +queries with 'P','I' replies laid out the way the provider parses them. A +headset that is switched off is a receiver that accepts writes and never +answers. time.sleep is replaced, so the tests count reads and writes rather +than wall-clock 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 barracuda as B # noqa: E402 + +PATH = b"\\\\?\\hid#vid_1532&pid_053a&mi_03&col01#x" + + +def reply(cmd, value): + r = [0x00] * 64 + r[0], r[1], r[2] = 0x01, 0x80, 0x0A + r[3], r[4] = B.MARKER_REPLY + r[5] = 0x08 + r[13], r[14], r[15], r[16] = cmd, 0x01, 1, value + return r + + +class FakeReceiver: + """answers: {cmd: value} for a headset that is on, None when it is off. + write_results: what write() returns, one per call (then len(frame)).""" + + def __init__(self, answers=None, write_results=()): + self.answers = answers + self.write_results = list(write_results) + self.written = [] + self.reads = 0 + self.pending = [] + + def device_class(self): + rx = self + + class FakeDevice: + def open_path(self, path): + pass + + def write(self, frame): + rx.written.append(bytes(frame)) + if rx.write_results: + return rx.write_results.pop(0) + if rx.answers is not None and frame[7] == B.TYPE_QUERY and frame[8] in rx.answers: + rx.pending.append(reply(frame[8], rx.answers[frame[8]])) + return len(frame) + + def read(self, n, timeout=0): + rx.reads += 1 + return rx.pending.pop(0) if rx.pending else [] + + def close(self): + pass + return FakeDevice + + def queries(self, cmd): + return sum(1 for f in self.written if f[7] == B.TYPE_QUERY and f[8] == cmd) + + def remotes(self): + return [f[9] for f in self.written if f[7] == B.TYPE_REMOTE and f[8] == B.CMD_REMOTE] + + +class ReadBatteryTest(unittest.TestCase): + def setUp(self): + self._saved = (B.hid, B.time) + self.sleeps = [] + B.time = types.SimpleNamespace(sleep=self.sleeps.append) + + def tearDown(self): + B.hid, B.time = self._saved + + def run_read(self, rx): + B.hid = types.SimpleNamespace(device=rx.device_class()) + diag = [] + return B.read_battery(PATH, diag), diag + + def test_headset_on(self): + rx = FakeReceiver({B.CMD_BATTERY: 77, B.CMD_CHARGING: 1}) + (res, _) = self.run_read(rx) + self.assertEqual(res, ("ok", 77, True)) + self.assertEqual(rx.remotes(), [1, 0]) + + def test_headset_off_skips_the_charging_query(self): + rx = FakeReceiver(None) + (res, diag) = self.run_read(rx) + self.assertEqual(res, ("offline", None, False)) + self.assertIn(" no battery reply", diag) + self.assertEqual(rx.queries(B.CMD_BATTERY), B.ATTEMPTS) + self.assertEqual(rx.queries(B.CMD_CHARGING), 0) + # remote mode is still switched off again, as before + self.assertEqual(rx.remotes(), [1, 0]) + + def test_headset_off_costs_one_query_of_reads(self): + rx = FakeReceiver(None) + self.run_read(rx) + # one query = ATTEMPTS x (1 drain read + 6 reads of REPLY_TIMEOUT_MS), plus one + # drain read for each of the two remote-mode frames; no second query + self.assertEqual(rx.reads, B.ATTEMPTS * (1 + 6) + 2) + + def test_failed_write_is_retried_for_remote_mode(self): + # hidapi returns -1 from write() instead of raising + rx = FakeReceiver({B.CMD_BATTERY: 50, B.CMD_CHARGING: 0}, write_results=[-1]) + (res, diag) = self.run_read(rx) + self.assertEqual(res, ("ok", 50, False)) + self.assertEqual(rx.remotes(), [1, 1, 0]) + self.assertEqual(self.sleeps[0], 0.15) + self.assertTrue(any("write" in line for line in diag)) + + def test_receiver_that_rejects_every_write_sends_no_queries(self): + rx = FakeReceiver({B.CMD_BATTERY: 50}, write_results=[-1] * B.WAKE_ATTEMPTS) + (res, diag) = self.run_read(rx) + self.assertEqual(res, ("offline", None, False)) + self.assertIn(" receiver does not accept commands", diag) + self.assertEqual(rx.queries(B.CMD_BATTERY), 0) + self.assertEqual(rx.remotes(), [1] * B.WAKE_ATTEMPTS) + + def test_failed_query_write_stops_that_query(self): + # remote mode on succeeds, then the battery query write fails + rx = FakeReceiver(None, write_results=[len(B.frame_remote(True)), -1]) + (res, _) = self.run_read(rx) + self.assertEqual(res, ("offline", None, False)) + self.assertEqual(rx.queries(B.CMD_BATTERY), 1) + + +class PollTest(unittest.TestCase): + def setUp(self): + from providers import hidlist + self._saved = (B.hid, hidlist.enumerate, B.time) + B.time = types.SimpleNamespace(sleep=lambda s: None) + hidlist.enumerate = lambda vid=0: [{"product_id": B.PID, "usage_page": B.USAGE_PAGE_VENDOR, + "usage": 1, "path": PATH}] + + def tearDown(self): + from providers import hidlist + B.hid, hidlist.enumerate, B.time = self._saved + + def test_headset_off_shows_the_offline_icon(self): + rx = FakeReceiver(None) + B.hid = types.SimpleNamespace(device=rx.device_class()) + res = B.BarracudaProvider().poll() + self.assertEqual([(r.key, r.name, r.level, r.charging, r.online, r.source, r.kind) for r in res], + [("barracuda:053a", "Razer Barracuda Pro (2.4 GHz)", None, False, False, + "barracuda", "headset")]) + + +if __name__ == "__main__": + unittest.main()