Jeff Leung
053897c287
Move hardware results into the sections implementers follow: history counts are 4-byte units, frames can be 160 bytes, transfers are padded and end with a length field, and status and marks are raw values. Add real frame vectors and the capture replay format so other languages can test. Change history_count to request units, with a margin for stop records. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
260 lines
9.5 KiB
Python
260 lines
9.5 KiB
Python
"""ANCEL BM200 (BM6 protocol family), from the com.dc.bm6_ancel 2.0.6 APK.
|
|
|
|
BM200Decoder is transport-free: feed it raw FFF4 notifications from any relay.
|
|
BM200Client does the GATT session (bleak needed only for this part).
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import struct
|
|
import time
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, List, Optional, Type, TypeVar, Union
|
|
|
|
from bm2.encryption import decrypt, encrypt
|
|
|
|
logger = logging.getLogger("bm200")
|
|
|
|
BM200_KEY = b"leagend\xff\xfe0100009"
|
|
UUID_NOTIFY = "0000fff4-0000-1000-8000-00805f9b34fb"
|
|
UUID_WRITE = "0000fff3-0000-1000-8000-00805f9b34fb"
|
|
|
|
CMD_VERSION = bytes.fromhex("d15501")
|
|
CMD_ACCEPT = bytes.fromhex("d1550200") # never send d1550211: the app's "not genuine" verdict
|
|
CMD_LIVE = bytes.fromhex("d15507")
|
|
|
|
RECORD_INTERVAL = 120 # seconds between history records
|
|
MAX_RECORDS_31D = 22320 # app's cap; it uses 51840 (72 days) only for "bm6intact" devices
|
|
|
|
|
|
def history_command(count: int) -> bytes:
|
|
return bytes.fromhex("d15505000000") + struct.pack(">L", count)[1:]
|
|
|
|
|
|
@dataclass
|
|
class Version:
|
|
version: float
|
|
|
|
|
|
@dataclass
|
|
class AcceptAck:
|
|
result: int
|
|
|
|
|
|
@dataclass
|
|
class LiveReading:
|
|
time: datetime
|
|
voltage: float
|
|
soc: int
|
|
temperature: int
|
|
status: int # raw; 2 = engine running (app treats 2 -> other as engine off)
|
|
rapid_acceleration: int
|
|
rapid_deceleration: int
|
|
|
|
|
|
@dataclass
|
|
class HistoryRecord:
|
|
time: datetime
|
|
voltage: float
|
|
soc: int
|
|
temperature: int
|
|
mark: Optional[str] # "start" / "stop" of the engine
|
|
|
|
|
|
@dataclass
|
|
class HistoryBlock:
|
|
records: List[HistoryRecord] # oldest first
|
|
|
|
|
|
Event = Union[Version, AcceptAck, LiveReading, HistoryBlock]
|
|
|
|
|
|
def _history_base(received_at: float) -> float:
|
|
# Same as the app: floor to the minute, then step back onto an odd local minute.
|
|
base = int(received_at) // 60 * 60
|
|
return base - 60 if datetime.fromtimestamp(base).minute % 2 == 0 else base
|
|
|
|
|
|
def parse_history(data: str, base: float) -> List[HistoryRecord]:
|
|
"""data: hex up to the FFFEFE marker. Record 0 is the newest, 2 minutes apart."""
|
|
# ponytail: strips all leading zero records; the app's loop also skips some records after them
|
|
while data.startswith("00000000"):
|
|
data = data[8:]
|
|
|
|
records = []
|
|
pos = index = 0
|
|
while pos + 8 <= len(data):
|
|
item = data[pos:pos + 8]
|
|
pos += 8
|
|
flags = int(item[7], 16)
|
|
mark = {0b010: "start", 0b100: "stop"}.get(flags & 0b111)
|
|
if mark == "stop":
|
|
pos += 8 # stop records carry 4 more bytes: rapid acceleration/deceleration counts
|
|
voltage = int(item[:3], 16) / 100
|
|
if voltage != 0: # zero-voltage slots are dropped but still take up a time slot
|
|
temperature = int(item[5:7], 16)
|
|
records.append(HistoryRecord(time=datetime.fromtimestamp(base - index * RECORD_INTERVAL, timezone.utc),
|
|
voltage=voltage,
|
|
soc=int(item[3:5], 16),
|
|
temperature=-temperature if flags & 0b1000 else temperature,
|
|
mark=mark))
|
|
index += 1
|
|
records.reverse()
|
|
return records
|
|
|
|
|
|
class BM200Decoder:
|
|
def __init__(self) -> None:
|
|
self._history: Optional[str] = None
|
|
self._history_base = 0.0
|
|
|
|
def feed(self, frame: Union[bytes, bytearray], received_at: Optional[float] = None) -> List[Event]:
|
|
"""Raises ValueError if frame is not whole AES blocks."""
|
|
if not frame or len(frame) % 16:
|
|
raise ValueError(f"bad frame length {len(frame)}")
|
|
now = time.time() if received_at is None else received_at
|
|
h = decrypt(frame, BM200_KEY).hex().upper()
|
|
|
|
# Live frames interleave with history and must not end up in it.
|
|
if h.startswith("D15507"):
|
|
return [] if h.startswith("D15507FF") else [self._live(h, now)]
|
|
|
|
if self._history is not None:
|
|
return self._history_frame(h)
|
|
|
|
if h.startswith("D15501"):
|
|
return [Version(int(h[6:8], 16) + int(h[8:10], 16) / 10)]
|
|
if h.startswith("D15502"):
|
|
return [AcceptAck(int(h[6:8], 16))]
|
|
if h.startswith("D15505"):
|
|
if int(h[6:10], 16) == 0:
|
|
return [HistoryBlock([])]
|
|
self._history = ""
|
|
self._history_base = _history_base(now)
|
|
return []
|
|
|
|
logger.debug("ignored frame %s", h)
|
|
return []
|
|
|
|
def _history_frame(self, h: str) -> List[Event]:
|
|
assert self._history is not None
|
|
if "FEFEFE" in h: # app: "empty history marker, discard"
|
|
return []
|
|
self._history += h
|
|
end = self._history.find("FFFEFE")
|
|
if end < 0:
|
|
return []
|
|
data, self._history = self._history[:end], None
|
|
return [HistoryBlock(parse_history(data, self._history_base))]
|
|
|
|
@staticmethod
|
|
def _live(h: str, now: float) -> LiveReading:
|
|
temperature = int(h[8:10], 16)
|
|
return LiveReading(time=datetime.fromtimestamp(now, timezone.utc),
|
|
voltage=int(h[14:18], 16) / 100,
|
|
soc=int(h[12:14], 16),
|
|
temperature=-temperature if int(h[6:8], 16) == 1 else temperature,
|
|
status=int(h[10:12], 16),
|
|
rapid_acceleration=int(h[18:22], 16),
|
|
rapid_deceleration=int(h[22:26], 16))
|
|
|
|
|
|
T = TypeVar("T")
|
|
|
|
|
|
class BM200Client:
|
|
def __init__(self, mac: str, since: Optional[datetime] = None, max_records: int = MAX_RECORDS_31D,
|
|
history_timeout: float = 600, live_timeout: float = 120):
|
|
self._mac = mac
|
|
self.since = since # newest history record delivered so far
|
|
self._max_records = max_records
|
|
self._history_timeout = history_timeout
|
|
self._live_timeout = live_timeout
|
|
|
|
def history_count(self) -> int:
|
|
if self.since is None:
|
|
return self._max_records
|
|
missing = max(int((datetime.now(timezone.utc) - self.since).total_seconds() / RECORD_INTERVAL), 0)
|
|
# In 4-byte units: stop records use two, so add a margin. +10 overlap is from the app.
|
|
return min(missing + missing // 50 + 10, self._max_records)
|
|
|
|
async def run(self,
|
|
on_reading: Callable[[Union[LiveReading, HistoryRecord]], None],
|
|
on_history_done: Optional[Callable[[datetime], None]] = None,
|
|
on_frame: Optional[Callable[[bytes], None]] = None) -> None:
|
|
"""Connect forever: on each connection backfill records newer than since, then stream live."""
|
|
import bleak.exc
|
|
from bleak import BleakClient
|
|
|
|
while True:
|
|
try:
|
|
async with BleakClient(self._mac) as client:
|
|
logger.info("connected")
|
|
await self._session(client, on_reading, on_history_done, on_frame)
|
|
except (bleak.exc.BleakError, OSError, asyncio.TimeoutError) as e:
|
|
logger.warning("session ended: %r", e)
|
|
await asyncio.sleep(5)
|
|
|
|
async def _session(self, client: Any, on_reading: Callable[..., None],
|
|
on_history_done: Optional[Callable[[datetime], None]],
|
|
on_frame: Optional[Callable[[bytes], None]]) -> None:
|
|
decoder = BM200Decoder()
|
|
queue: "asyncio.Queue[Event]" = asyncio.Queue()
|
|
last_frame = time.monotonic()
|
|
|
|
def handler(_: Any, data: bytearray) -> None:
|
|
nonlocal last_frame
|
|
last_frame = time.monotonic()
|
|
if on_frame:
|
|
on_frame(bytes(data))
|
|
try:
|
|
events = decoder.feed(data)
|
|
except ValueError as e:
|
|
logger.warning("%s", e)
|
|
return
|
|
for event in events:
|
|
if isinstance(event, LiveReading):
|
|
on_reading(event)
|
|
else:
|
|
queue.put_nowait(event)
|
|
|
|
async def send(cmd: bytes) -> None:
|
|
await client.write_gatt_char(UUID_WRITE, encrypt(cmd, BM200_KEY))
|
|
|
|
async def expect(kind: Type[T], timeout: float = 10) -> T:
|
|
while True:
|
|
event = await asyncio.wait_for(queue.get(), timeout)
|
|
if isinstance(event, kind):
|
|
return event
|
|
|
|
await client.start_notify(UUID_NOTIFY, handler)
|
|
|
|
# Same order as the app without its cloud check. Its battery-type write (D15508) is
|
|
# deliberately left out: it changes tag configuration.
|
|
await send(CMD_VERSION)
|
|
logger.info("firmware %s", (await expect(Version)).version)
|
|
await send(CMD_ACCEPT)
|
|
await expect(AcceptAck)
|
|
|
|
# The app always reads and discards 10 records before the real read.
|
|
await send(history_command(10))
|
|
await expect(HistoryBlock)
|
|
|
|
await send(history_command(self.history_count()))
|
|
block = await expect(HistoryBlock, self._history_timeout)
|
|
for record in block.records:
|
|
if self.since is None or record.time > self.since:
|
|
on_reading(record)
|
|
self.since = record.time
|
|
if self.since is not None and on_history_done:
|
|
on_history_done(self.since)
|
|
|
|
await send(CMD_LIVE)
|
|
while client.is_connected:
|
|
if time.monotonic() - last_frame > self._live_timeout:
|
|
raise asyncio.TimeoutError("no frames")
|
|
await asyncio.sleep(1)
|
|
|
|
|
|
__all__ = ["BM200Client", "BM200Decoder", "HistoryBlock", "HistoryRecord", "LiveReading", "parse_history"]
|