bm200/findings/reference-python/bm200.py

"""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 = (datetime.now(timezone.utc) - self.since).total_seconds() / RECORD_INTERVAL
        return min(max(int(missing), 0) + 10, self._max_records)  # +10 overlap, as in the app

    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"]