Crack the frame checksum; generate frames programmatically

Recovered the protocol's checksum from Gadgetbridge: CRC-16/MODBUS
(poly 0xA001, init 0xFFFF, reflected, little-endian) over the whole
frame minus the trailing two CRC bytes. The frame format is
0x55 | control | command | length | fsn | payload | crc.

Verified by reproducing three independent known-good frames
byte-for-byte (ANC Off, Transparency, status query).

- protocol.py: add crc16ansi() + build_frame(); generate ANC, EQ and
  query frames from semantic fields instead of hardcoded hex (some of
  the previously hardcoded CRCs were transcription-corrupted).
- Controller now uses a frame sequence number per command.
- docs/PROTOCOL.md: document the solved frame format and checksum.

https://claude.ai/code/session_014KrpftzUgVFGFXVUU3DUTT

Author: claude

docs/PROTOCOL.md

@@ -22,6 +22,32 @@ ANC frames below are verified to work on a real Nothing Ear (3).**
 | Socket (Linux)  | `socket(AF_BLUETOOTH, SOCK_STREAM, BTPROTO_RFCOMM)` → `connect((mac, 15))` |
 | Battery         | available directly via BlueZ/`upower` (no protocol needed) — already implemented |
 
+### Frame format and checksum (solved)
+
+```
+0x55 | control(2, LE) | command(2, LE) | length(2, LE) | fsn(1) | payload | crc(2, LE)
+```
+
+- `control` `0x0160` sets the increment-counter and CRC-present flags.
+- `fsn` is a frame sequence number (this is the byte that "randomly" varied
+  across captured frames — it is not data).
+- `crc` is **CRC-16/MODBUS** (poly `0xA001`, init `0xFFFF`, reflected) over the
+  whole frame except the trailing two CRC bytes. Verified: it reproduces three
+  independent known-good frames byte-for-byte (ANC Off, ANC Transparency, ANC
+  status query), and the MODBUS residue over a full frame is `0`.
+
+This is implemented as `crc16ansi()` + `build_frame()` in
+[`../src/nothing_ear/bluetooth/protocol.py`](../src/nothing_ear/bluetooth/protocol.py),
+so frames are now generated programmatically instead of hardcoded.
+
+Source of the algorithm: Gadgetbridge `CheckSums.getCRC16ansi` and
+`NothingProtocol`. Note: the real Ear (3) appears to **not strictly validate**
+the CRC (frames with a wrong CRC still took effect), but we compute it correctly
+anyway for compatibility.
+
+Commands: `0xf00f` = ANC set, `0xc01e` = ANC status query, `0x4006` = EQ preset
+(EQ command/payload is a reconstruction, pending hardware confirmation).
+
 ### Known ANC frames (13-byte frames)
 
 Structure: `55 60 01 0f f0 03 00 <a> 01 <mode> 00 <crc16>`

src/nothing_ear/bluetooth/protocol.py

@@ -1,27 +1,34 @@
 """Nothing control protocol over Bluetooth Classic RFCOMM.
 
-The command frames were reverse-engineered by the community and verified to
-work on the Nothing Ear (3).
-
-Protocol summary:
-- Transport: Bluetooth Classic, RFCOMM (SPP), channel 15.
-- Every frame starts with the preamble byte 0x55.
-- ANC "set" frames are 13 bytes; byte index 9 selects the mode
-  (01=High, 02=Mid, 03=Low, 04=Adaptive, 05=Off). The trailing two bytes are
-  a CRC-16, so we ship full precomputed frames.
-
-Sources: bharadwaj-raju.github.io/posts/nothing-ear-2-on-linux,
-LuanAdemi/nothing-ear-controller, OPIYOdev/Nothing-Ear-Linux, DaanHessen/earctl.
+Frame format (verified against Gadgetbridge and real Ear (3) frames):
+
+    0x55 | control(2, LE) | command(2, LE) | length(2, LE) | fsn(1) | payload | crc(2, LE)
+
+- control 0x0160 sets the "increment counter" and "CRC present" flags.
+- fsn is a frame sequence number.
+- crc is CRC-16/MODBUS (poly 0xA001, init 0xFFFF, reflected) over the whole
+  frame except the trailing two CRC bytes.
+
+Transport: Bluetooth Classic, RFCOMM (SPP), channel 15.
+
+Sources: Freeyourgadget/Gadgetbridge (NothingProtocol, CheckSums.getCRC16ansi),
+plus frames verified on real Ear (3) hardware.
 """
 
 from __future__ import annotations
 
 import socket
+import struct
 import time
 from enum import Enum
 
 RFCOMM_CHANNEL = 15
 PACKET_PREAMBLE = 0x55
+DEFAULT_CONTROL = 0x0160
+
+CMD_ANC = 0xF00F
+CMD_ANC_STATUS = 0xC01E
+CMD_EQ = 0x4006
 
 
 class ANCMode(Enum):
@@ -40,33 +47,64 @@ class EQPreset(Enum):
     VOICE = "voice"
 
 
-ANC_COMMANDS: dict[ANCMode, str] = {
-    ANCMode.HIGH: "5560010ff00300cf010100e66f",
-    ANCMode.MID: "5560010ff00300d5010200e69f",
-    ANCMode.LOW: "5560010ff00300d7010300e70f",
-    ANCMode.ADAPTIVE: "5560010ff00300dd010400e53f",
-    ANCMode.OFF: "5560010ff00300cd010500c447",
-    ANCMode.TRANSPARENCY: "5560010ff00300cb010700c5af",
+_ANC_VALUE = {
+    ANCMode.HIGH: 0x01,
+    ANCMode.MID: 0x02,
+    ANCMode.LOW: 0x03,
+    ANCMode.ADAPTIVE: 0x04,
+    ANCMode.OFF: 0x05,
+    ANCMode.TRANSPARENCY: 0x07,
 }
+_ANC_BY_VALUE = {v: k for k, v in _ANC_VALUE.items()}
 
-EQ_COMMANDS: dict[EQPreset, str] = {
-    EQPreset.BALANCED: "55600106400000060101009f3c",
-    EQPreset.MORE_BASS: "55600106400000060101019e8c",
-    EQPreset.MORE_TREBLE: "55600106400000060101029e1c",
-    EQPreset.VOICE: "55600106400000060101039eac",
+_EQ_VALUE = {
+    EQPreset.BALANCED: 0x00,
+    EQPreset.MORE_BASS: 0x01,
+    EQPreset.MORE_TREBLE: 0x02,
+    EQPreset.VOICE: 0x03,
 }
 
-QUERY_ANC = "5560011ec001000c039819"
-
 ANC_RESPONSE_MODE_INDEX = 9
-_MODE_BY_VALUE = {
-    1: ANCMode.HIGH,
-    2: ANCMode.MID,
-    3: ANCMode.LOW,
-    4: ANCMode.ADAPTIVE,
-    5: ANCMode.OFF,
-    7: ANCMode.TRANSPARENCY,
-}
+
+
+def crc16ansi(data: bytes) -> int:
+    """CRC-16/MODBUS (poly 0xA001, init 0xFFFF, reflected)."""
+    crc = 0xFFFF
+    for byte in data:
+        crc ^= byte & 0xFF
+        for _ in range(8):
+            crc = (crc >> 1) ^ 0xA001 if crc & 1 else crc >> 1
+    return crc & 0xFFFF
+
+
+def build_frame(command: int, payload: bytes, fsn: int = 0, control: int = DEFAULT_CONTROL) -> bytes:
+    """Assemble a full control frame including the trailing CRC."""
+    body = (
+        bytes([PACKET_PREAMBLE])
+        + struct.pack("<H", control)
+        + struct.pack("<H", command)
+        + struct.pack("<H", len(payload))
+        + bytes([fsn & 0xFF])
+        + payload
+    )
+    return body + struct.pack("<H", crc16ansi(body))
+
+
+def anc_frame(mode: ANCMode, fsn: int = 0) -> bytes:
+    return build_frame(CMD_ANC, bytes([0x01, _ANC_VALUE[mode], 0x00]), fsn)
+
+
+def eq_preset_frame(preset: EQPreset, fsn: int = 0) -> bytes:
+    return build_frame(CMD_EQ, bytes([0x01, 0x01, _EQ_VALUE[preset]]), fsn)
+
+
+def query_anc_frame(fsn: int = 0) -> bytes:
+    return build_frame(CMD_ANC_STATUS, bytes([0x03]), fsn)
+
+
+ANC_COMMANDS = {mode: anc_frame(mode).hex() for mode in ANCMode}
+EQ_COMMANDS = {preset: eq_preset_frame(preset).hex() for preset in EQPreset}
+QUERY_ANC = query_anc_frame().hex()
 
 
 class NothingController:
@@ -82,6 +120,7 @@ def __init__(self, address: str, channel: int = RFCOMM_CHANNEL):
         self.address = address
         self.channel = channel
         self._sock: socket.socket | None = None
+        self._fsn = 0
 
     def connect(self) -> None:
         sock = socket.socket(
@@ -104,23 +143,34 @@ def __enter__(self) -> "NothingController":
     def __exit__(self, *exc) -> None:
         self.close()
 
+    def _next_fsn(self) -> int:
+        fsn = self._fsn
+        self._fsn = (self._fsn + 1) & 0xFF
+        return fsn
+
     def send_raw(self, hex_frame: str) -> None:
         if self._sock is None:
             raise RuntimeError("Not connected. Call connect() first.")
         self._sock.sendall(bytes.fromhex(hex_frame))
         time.sleep(0.3)
 
+    def send_frame(self, frame: bytes) -> None:
+        if self._sock is None:
+            raise RuntimeError("Not connected. Call connect() first.")
+        self._sock.sendall(frame)
+        time.sleep(0.3)
+
     def set_anc(self, mode: ANCMode) -> None:
-        self.send_raw(ANC_COMMANDS[mode])
+        self.send_frame(anc_frame(mode, self._next_fsn()))
 
     def set_eq(self, preset: EQPreset) -> None:
-        self.send_raw(EQ_COMMANDS[preset])
+        self.send_frame(eq_preset_frame(preset, self._next_fsn()))
 
     def query_anc(self) -> ANCMode | None:
         if self._sock is None:
             raise RuntimeError("Not connected. Call connect() first.")
-        self._sock.sendall(bytes.fromhex(QUERY_ANC))
+        self._sock.sendall(query_anc_frame(self._next_fsn()))
         resp = self._sock.recv(64)
         if len(resp) > ANC_RESPONSE_MODE_INDEX:
-            return _MODE_BY_VALUE.get(resp[ANC_RESPONSE_MODE_INDEX])
+            return _ANC_BY_VALUE.get(resp[ANC_RESPONSE_MODE_INDEX])
         return None