ble version

collector/protocol.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+"""Pack/unpack 4-byte LE sensor reports and slot/config replies."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional, Tuple
+
+BATT_UNKNOWN = 0xFF
+DIST_NO_TARGET = 0xFFFF
+SLOT_UNASSIGNED = 0xFF
+
+FLAG_LOW_BATT = 0x01
+FLAG_FAULT = 0x02
+FLAG_NO_TARGET = 0x04
+FLAG_UNCHANGED = 0x08
+
+CFG_JOIN_OPEN = 0x01
+CFG_POOL_FULL = 0x02
+CFG_BUSY = 0x04
+
+
+@dataclass(frozen=True)
+class SensorReport:
+    battery: int
+    flags: int
+    dist: int
+
+
+@dataclass(frozen=True)
+class SlotConfig:
+    slot: int
+    flags: int
+    period_s: int
+
+
+def unpack_report(data: bytes) -> Optional[SensorReport]:
+    if data is None or len(data) != 4:
+        return None
+    battery = data[0]
+    flags = data[1]
+    dist = data[2] | (data[3] << 8)
+    return SensorReport(battery=battery, flags=flags, dist=dist)
+
+
+def pack_report(battery: int, flags: int, dist: int) -> bytes:
+    batt = max(0, min(255, int(battery))) & 0xFF
+    fl = int(flags) & 0xFF
+    d = max(0, min(0xFFFF, int(dist))) & 0xFFFF
+    return bytes((batt, fl, d & 0xFF, (d >> 8) & 0xFF))
+
+
+def unpack_slot_config(data: bytes) -> Optional[SlotConfig]:
+    if data is None or len(data) != 4:
+        return None
+    slot = data[0]
+    flags = data[1]
+    period = data[2] | (data[3] << 8)
+    return SlotConfig(slot=slot, flags=flags, period_s=period)
+
+
+def pack_slot_config(slot: int, flags: int, period_s: int) -> bytes:
+    s = max(0, min(255, int(slot))) & 0xFF
+    fl = int(flags) & 0xFF
+    p = max(0, min(0xFFFF, int(period_s))) & 0xFFFF
+    return bytes((s, fl, p & 0xFF, (p >> 8) & 0xFF))
+
+
+def report_to_bytes(r: SensorReport) -> bytes:
+    return pack_report(r.battery, r.flags, r.dist)