123 lines
3.0 KiB
Python
123 lines
3.0 KiB
Python
#!/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)
|
||
|
||
|
||
def pack_interval_min(minutes: int) -> bytes:
|
||
"""2-byte LE uint16 report interval in minutes."""
|
||
m = max(1, min(0xFFFF, int(minutes))) & 0xFFFF
|
||
return bytes((m & 0xFF, (m >> 8) & 0xFF))
|
||
|
||
|
||
def unpack_interval_min(data: bytes) -> Optional[int]:
|
||
if data is None or len(data) != 2:
|
||
return None
|
||
return data[0] | (data[1] << 8)
|
||
|
||
|
||
# 2×AA alkaline, light load. Knots are (millivolts, percent).
|
||
# Below the first knot is 0%, above the last is 100%.
|
||
_CHARGE_CURVE_MV = (
|
||
(1700, 0),
|
||
(2000, 8),
|
||
(2200, 20),
|
||
(2400, 35),
|
||
(2600, 52),
|
||
(2800, 72),
|
||
(3000, 90),
|
||
(3200, 100),
|
||
)
|
||
|
||
|
||
def voltage_v(byte: int) -> Optional[float]:
|
||
"""Byte 0 is tenths of a volt. 255 means no reading."""
|
||
if byte is None:
|
||
return None
|
||
b = int(byte) & 0xFF
|
||
if b == BATT_UNKNOWN:
|
||
return None
|
||
return b / 10.0
|
||
|
||
|
||
def charge_percent(volts: float) -> int:
|
||
"""Approximate remaining charge for two cheap alkaline AA cells."""
|
||
mv = int(round(float(volts) * 1000))
|
||
if mv <= _CHARGE_CURVE_MV[0][0]:
|
||
return 0
|
||
if mv >= _CHARGE_CURVE_MV[-1][0]:
|
||
return 100
|
||
for (v0, p0), (v1, p1) in zip(_CHARGE_CURVE_MV, _CHARGE_CURVE_MV[1:]):
|
||
if mv <= v1:
|
||
den = v1 - v0
|
||
if den <= 0:
|
||
return int(p1)
|
||
num = (mv - v0) * (p1 - p0)
|
||
return int(p0 + (num + den // 2) // den)
|
||
return 100
|