Add BullTron BLE desktop GUI

.gitignore

@@ -6,3 +6,6 @@ apk/
 *.dex
 *.jar
 *.class
+__pycache__/
+*.py[cod]
+.venv/

README.md

@@ -52,6 +52,43 @@ Send that frame to characteristic `0000fff2-0000-1000-8000-00805f9b34fb`
 after subscribing to notifications on
 `0000fff1-0000-1000-8000-00805f9b34fb`.
 
+## Local Desktop App
+
+This repo includes a small Python GUI client that can scan for BullTron-style
+BLE devices, connect, poll telemetry, and display the decoded values locally on
+a PC.
+
+Install:
+
+```sh
+python3 -m venv .venv
+. .venv/bin/activate
+python3 -m pip install -r requirements.txt
+```
+
+Run:
+
+```sh
+python3 bulltron_gui.py
+```
+
+The GUI shows:
+
+- live pack voltage, current, discharge watts, SOC, remaining Ah, cell max/min,
+  imbalance, cell count, cycle count, charge MOS, discharge MOS, and computed
+  time-to-empty/time-to-full
+- alarm/status words decoded into readable alarm names where known
+- system/settings values such as the control PIN, firmware/version text,
+  product info, battery code/SN, production-date raw value, and inferred battery
+  Ah when the settings block contains a plausible capacity
+- raw TX/RX frames for debugging and protocol confirmation
+
+The settings tab has MOS on/off controls, but writes are deliberately guarded:
+you must enter the BMS PIN, tick `Enable MOS writes`, and confirm each write.
+The default write registers are the observed live MOS status registers
+`0x0035` and `0x0036`; keep them editable until those write registers are
+verified on your exact hardware/firmware.
+
 ## App Workflow
 
 The app's normal telemetry flow is:

bulltron_gui.py

@@ -0,0 +1,612 @@
+#!/usr/bin/env python3
+"""Tkinter BLE dashboard for BullTron batteries."""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import queue
+import threading
+import time
+from dataclasses import dataclass, field
+from typing import Any, Callable
+
+try:
+    import tkinter as tk
+    from tkinter import messagebox, ttk
+except ImportError:  # pragma: no cover - depends on system Tk packages.
+    class _MissingTk:
+        class Tk:
+            pass
+
+        class StringVar:
+            def __init__(self, value: str = "") -> None:
+                self.value = value
+
+            def get(self) -> str:
+                return self.value
+
+            def set(self, value: str) -> None:
+                self.value = value
+
+        class BooleanVar(StringVar):
+            pass
+
+        END = "end"
+        BOTH = "both"
+        X = "x"
+        HORIZONTAL = "horizontal"
+        LEFT = "left"
+
+    tk = _MissingTk()
+    messagebox = None
+    ttk = None
+
+try:
+    from bleak import BleakClient, BleakScanner
+except ImportError:  # pragma: no cover - exercised by users without deps.
+    BleakClient = None
+    BleakScanner = None
+
+
+MAIN_SERVICE = "0000fff0-0000-1000-8000-00805f9b34fb"
+NOTIFY_CHAR = "0000fff1-0000-1000-8000-00805f9b34fb"
+WRITE_CHAR = "0000fff2-0000-1000-8000-00805f9b34fb"
+AT_CHAR = "0000fff3-0000-1000-8000-00805f9b34fb"
+SECRET_KEY_CHAR = "02f00000-0000-0000-0000-00000000ff05"
+VERSION_CHAR = "02f00000-0000-0000-0000-00000000ff04"
+
+
+def crc16_modbus_swapped(data: bytes) -> int:
+    crc = 0xFFFF
+    for byte in data:
+        crc ^= byte
+        for _ in range(8):
+            if crc & 1:
+                crc = (crc >> 1) ^ 0xA001
+            else:
+                crc >>= 1
+    return ((crc & 0xFF00) >> 8) | ((crc & 0x00FF) << 8)
+
+
+def read_frame(start: int, count: int) -> bytes:
+    body = bytes([0xD2, 0x03]) + start.to_bytes(2, "big") + count.to_bytes(2, "big")
+    return body + crc16_modbus_swapped(body).to_bytes(2, "big")
+
+
+def write_single_frame(register: int, value: int) -> bytes:
+    body = bytes([0xD2, 0x06]) + register.to_bytes(2, "big") + value.to_bytes(2, "big")
+    return body + crc16_modbus_swapped(body).to_bytes(2, "big")
+
+
+def write_time_frame() -> bytes:
+    now = time.localtime()
+    payload = bytes([now.tm_year - 2000, now.tm_mon, now.tm_mday, now.tm_hour, now.tm_min, now.tm_sec])
+    body = bytes([0xD2, 0x10, 0x00, 0xD4, 0x00, 0x03]) + payload
+    return body + crc16_modbus_swapped(body).to_bytes(2, "big")
+
+
+def words_from_payload(payload: bytes) -> list[int]:
+    return [int.from_bytes(payload[i : i + 2], "big") for i in range(0, len(payload) - 1, 2)]
+
+
+def printable_ascii(data: bytes) -> str:
+    text = "".join(chr(b) if 32 <= b <= 126 else "." for b in data)
+    return text.strip(".\x00 ")
+
+
+def format_duration(minutes: float | None) -> str:
+    if minutes is None or minutes <= 0:
+        return "-- h -- m"
+    hours, mins = divmod(int(round(minutes)), 60)
+    return f"{hours} h {mins:02d} m"
+
+
+def split_frames(buffer: bytearray) -> list[bytes]:
+    frames: list[bytes] = []
+    while True:
+        start = buffer.find(b"\xD2")
+        if start < 0:
+            buffer.clear()
+            return frames
+        if start:
+            del buffer[:start]
+        if len(buffer) < 3:
+            return frames
+        func = buffer[1]
+        if func == 0x03:
+            frame_len = 3 + buffer[2] + 2
+        elif func in (0x06, 0x10):
+            frame_len = 8
+        else:
+            del buffer[0]
+            continue
+        if len(buffer) < frame_len:
+            return frames
+        frame = bytes(buffer[:frame_len])
+        del buffer[:frame_len]
+        if len(frame) >= 5:
+            expected = int.from_bytes(frame[-2:], "big")
+            actual = crc16_modbus_swapped(frame[:-2])
+            if expected == actual:
+                frames.append(frame)
+
+
+@dataclass
+class Telemetry:
+    cells: list[float] = field(default_factory=list)
+    voltage: float | None = None
+    current: float | None = None
+    soc: float | None = None
+    remaining_ah: float | None = None
+    power_w: float | None = None
+    state: int | None = None
+    cycle_count: int | None = None
+    charge_mos: bool | None = None
+    discharge_mos: bool | None = None
+    cell_max: float | None = None
+    cell_min: float | None = None
+    cell_delta: float | None = None
+    cell_count: int | None = None
+    alarm_words: list[int] = field(default_factory=list)
+    time_label: str = "Time"
+    time_value: str = "-- h -- m"
+
+
+@dataclass
+class DeviceState:
+    telemetry: Telemetry = field(default_factory=Telemetry)
+    rated_capacity_ah: float | None = None
+    control_pin: str | None = None
+    firmware: str = ""
+    battery_code: str = ""
+    production_date: str = ""
+    product_info: str = ""
+    raw_blocks: dict[str, str] = field(default_factory=dict)
+    alarms: list[str] = field(default_factory=list)
+
+
+ALARM_LABELS = [
+    "Cell volt high level 1", "Cell volt high level 2", "Cell volt low level 1", "Cell volt low level 2",
+    "Sum volt high level 1", "Sum volt high level 2", "Sum volt low level 1", "Sum volt low level 2",
+    "Chg temp high level 1", "Chg temp high level 2", "Chg temp low level 1", "Chg temp low level 2",
+    "Dischg temp high level 1", "Dischg temp high level 2", "Dischg temp low level 1", "Dischg temp low level 2",
+    "Chg overcurrent level 1", "Chg overcurrent level 2", "Dischg overcurrent level 1", "Dischg overcurrent level 2",
+    "SOC high level 1", "SOC high level 2", "SOC Low level 1", "SOC Low level 2",
+    "Diff volt level 1", "Diff volt level 2", "Diff temp level 1", "Diff temp level 2",
+    "Short circuit protect fault", "AFE collect chip err", "Voltage collect dropped", "Cell temp sensor err",
+    "EEPROM err", "RTC err", "Precharge failure", "Communication failure",
+    "Internal communication failure", "Current module fault", "Sum voltage detect fault", "Low volt forbidden chg fault",
+    "Chg MOS temp high alarm", "Dischg MOS temp high alarm", "Chg MOS temp sensor err", "Dischg MOS temp sensor err",
+    "Chg MOS adhesion err", "Dischg MOS adhesion err", "Chg MOS open circuit err", "Discrg MOS open circuit err",
+]
+
+
+def decode_live(payload: bytes, state: DeviceState) -> None:
+    words = words_from_payload(payload)
+    if len(words) < 62:
+        return
+    tel = state.telemetry
+    cells = [raw * 0.001 for raw in words[:32] if raw > 0]
+    tel.cells = cells
+    tel.cell_count = len(cells)
+    tel.voltage = words[40] * 0.1
+    tel.current = (words[41] - 30000) * 0.1
+    tel.soc = words[42] / 10
+    tel.cell_max = words[43] * 0.001
+    tel.cell_min = words[44] * 0.001
+    tel.state = words[47]
+    tel.remaining_ah = words[48] * 0.1
+    tel.cycle_count = words[51]
+    tel.charge_mos = words[53] == 1
+    tel.discharge_mos = words[54] == 1
+    tel.cell_delta = words[56] * 0.001
+    tel.alarm_words = words[58:62]
+    if tel.voltage is not None and tel.current is not None:
+        tel.power_w = abs(tel.voltage * tel.current)
+    capacity = state.rated_capacity_ah or tel.remaining_ah
+    minutes = None
+    if capacity and tel.soc is not None and tel.current and abs(tel.current) > 0.05:
+        if tel.state == 1:
+            tel.time_label = "Time till full"
+            minutes = capacity * (100 - tel.soc) / abs(tel.current) * 60
+        elif tel.state == 2:
+            tel.time_label = "Time till empty"
+            minutes = capacity * tel.soc / abs(tel.current) * 60
+        else:
+            tel.time_label = "Time"
+    tel.time_value = format_duration(minutes)
+    state.alarms = decode_alarm_words(tel.alarm_words)
+
+
+def decode_alarm_words(words: list[int]) -> list[str]:
+    active: list[str] = []
+    for word_index, word in enumerate(words):
+        for bit in range(16):
+            if word & (1 << bit):
+                label_index = word_index * 16 + bit
+                label = ALARM_LABELS[label_index] if label_index < len(ALARM_LABELS) else f"Alarm word {word_index + 1} bit {bit}"
+                active.append(label)
+    return active
+
+
+def decode_response(frame: bytes, state: DeviceState) -> None:
+    if frame[1] != 0x03:
+        state.raw_blocks["Last ACK"] = frame.hex(" ").upper()
+        return
+    byte_count = frame[2]
+    payload = frame[3:-2]
+    state.raw_blocks[f"D203 byte_count 0x{byte_count:02X}"] = payload.hex(" ").upper()
+    if byte_count == 0x7C:
+        decode_live(payload, state)
+    elif byte_count == 0x06:
+        text = printable_ascii(payload)
+        state.control_pin = text or payload.hex().upper()
+    elif byte_count == 0x40:
+        state.firmware = printable_ascii(payload) or payload.hex().upper()
+    elif byte_count == 0x18:
+        state.battery_code = printable_ascii(payload) or payload.hex().upper()
+    elif byte_count == 0x04:
+        words = words_from_payload(payload)
+        if len(words) == 2:
+            state.production_date = f"raw {words[0]:04X} {words[1]:04X}"
+    elif byte_count == 0x52:
+        parse_settings(payload, state)
+
+
+def parse_settings(payload: bytes, state: DeviceState) -> None:
+    words = words_from_payload(payload)
+    candidates = []
+    for raw in words:
+        val = raw * 0.1
+        if 20 <= val <= 1000:
+            candidates.append(val)
+    if candidates and state.rated_capacity_ah is None:
+        state.rated_capacity_ah = max(candidates)
+
+
+class BleWorker:
+    def __init__(self, ui_queue: queue.Queue[tuple[str, Any]]) -> None:
+        self.ui_queue = ui_queue
+        self.loop = asyncio.new_event_loop()
+        self.thread = threading.Thread(target=self._run_loop, daemon=True)
+        self.thread.start()
+        self.client: Any = None
+        self.buffer = bytearray()
+        self.state = DeviceState()
+        self.polling = False
+
+    def _run_loop(self) -> None:
+        asyncio.set_event_loop(self.loop)
+        self.loop.run_forever()
+
+    def submit(self, coro: Any) -> None:
+        asyncio.run_coroutine_threadsafe(coro, self.loop)
+
+    def scan(self, timeout: float = 6.0) -> None:
+        self.submit(self._scan(timeout))
+
+    async def _scan(self, timeout: float) -> None:
+        if BleakScanner is None:
+            self.ui_queue.put(("error", "Install dependencies first: python3 -m pip install -r requirements.txt"))
+            return
+        self.ui_queue.put(("status", "Scanning..."))
+        devices = await BleakScanner.discover(timeout=timeout, return_adv=True)
+        rows = []
+        for _, (dev, adv) in devices.items():
+            name = dev.name or adv.local_name or "(unnamed)"
+            service_uuids = [u.lower() for u in (adv.service_uuids or [])]
+            likely = "bull" in name.lower() or MAIN_SERVICE in service_uuids
+            rows.append({"name": name, "address": dev.address, "rssi": adv.rssi, "likely": likely})
+        rows.sort(key=lambda row: (not row["likely"], -(row["rssi"] or -999)))
+        self.ui_queue.put(("scan", rows))
+        self.ui_queue.put(("status", f"Found {len(rows)} BLE devices"))
+
+    def connect(self, address: str) -> None:
+        self.submit(self._connect(address))
+
+    async def _connect(self, address: str) -> None:
+        if BleakClient is None:
+            self.ui_queue.put(("error", "Install dependencies first: python3 -m pip install -r requirements.txt"))
+            return
+        await self.disconnect()
+        self.ui_queue.put(("status", f"Connecting to {address}..."))
+        self.client = BleakClient(address, disconnected_callback=lambda _: self.ui_queue.put(("status", "Disconnected")))
+        await self.client.connect()
+        self.buffer.clear()
+        self.state = DeviceState()
+        try:
+            await self.client.start_notify(NOTIFY_CHAR, self._notification)
+        except Exception as exc:
+            self.ui_queue.put(("error", f"Could not enable fff1 notifications: {exc}"))
+            await self.disconnect()
+            return
+        for char in (AT_CHAR, VERSION_CHAR):
+            try:
+                await self.client.start_notify(char, lambda _sender, data, source=char: self._text_notification(source, data))
+            except Exception as exc:
+                self.ui_queue.put(("log", f"Optional notify on {char[-4:]} skipped: {exc}"))
+        await self._optional_write(SECRET_KEY_CHAR, b"HiLink", response=False)
+        await self._optional_write(VERSION_CHAR, b"AT+VER=?\r\n", response=False)
+        await self._optional_write(AT_CHAR, b"AT+PRODUCTINFO=?", response=False)
+        await self._write_command(write_time_frame())
+        self.polling = True
+        self.ui_queue.put(("connected", address))
+        self.submit(self._poll_loop())
+
+    async def _optional_write(self, char: str, data: bytes, response: bool) -> None:
+        try:
+            if self.client and self.client.is_connected:
+                await self.client.write_gatt_char(char, data, response=response)
+        except Exception as exc:
+            self.ui_queue.put(("log", f"Optional write to {char[-4:]} skipped: {exc}"))
+
+    async def _write_command(self, data: bytes) -> None:
+        if self.client and self.client.is_connected:
+            await self.client.write_gatt_char(WRITE_CHAR, data, response=False)
+            self.ui_queue.put(("tx", data.hex(" ").upper()))
+
+    async def _poll_loop(self) -> None:
+        slow_counter = 0
+        while self.polling and self.client and self.client.is_connected:
+            try:
+                await self._write_command(read_frame(0x0000, 0x003E))
+                slow_counter += 1
+                if slow_counter % 5 == 1:
+                    for start, count in ((0x003E, 0x0010), (0x0064, 0x0001), (0x0080, 0x0029), (0x00A9, 0x0025), (0x00C9, 0x0003), (0x00D9, 0x0005)):
+                        await asyncio.sleep(0.12)
+                        await self._write_command(read_frame(start, count))
+                await asyncio.sleep(2.0)
+            except Exception as exc:
+                self.ui_queue.put(("error", f"Polling failed: {exc}"))
+                await self.disconnect()
+                return
+
+    def _notification(self, _sender: Any, data: bytearray) -> None:
+        self.buffer.extend(bytes(data))
+        for frame in split_frames(self.buffer):
+            decode_response(frame, self.state)
+            self.ui_queue.put(("rx", frame.hex(" ").upper()))
+            self.ui_queue.put(("state", self.state))
+
+    def _text_notification(self, source: str, data: bytearray) -> None:
+        raw = bytes(data)
+        text = printable_ascii(raw) or raw.hex(" ").upper()
+        sender_text = str(source)
+        if "ff04" in sender_text.lower():
+            self.state.firmware = text
+        elif "fff3" in sender_text.lower():
+            if "PRODUCTINFO" in text.upper() or text:
+                self.state.product_info = text
+        self.ui_queue.put(("rx", f"{sender_text}: {raw.hex(' ').upper()}"))
+        self.ui_queue.put(("state", self.state))
+
+    def write_mos(self, register: int, enabled: bool) -> None:
+        self.submit(self._write_command(write_single_frame(register, 1 if enabled else 0)))
+
+    async def disconnect(self) -> None:
+        self.polling = False
+        if self.client:
+            try:
+                if self.client.is_connected:
+                    await self.client.disconnect()
+            finally:
+                self.client = None
+
+
+class BullTronGui(tk.Tk):
+    def __init__(self) -> None:
+        super().__init__()
+        self.title("BullTron BLE Dashboard")
+        self.geometry("1050x720")
+        self.queue: queue.Queue[tuple[str, Any]] = queue.Queue()
+        self.worker = BleWorker(self.queue)
+        self.devices: list[dict[str, Any]] = []
+        self.vars: dict[str, tk.StringVar] = {}
+        self.unlocked = False
+        self._build_ui()
+        self.after(100, self._drain_queue)
+
+    def _build_ui(self) -> None:
+        root = ttk.PanedWindow(self, orient=tk.HORIZONTAL)
+        root.pack(fill=tk.BOTH, expand=True, padx=8, pady=8)
+
+        left = ttk.Frame(root, width=300)
+        root.add(left, weight=0)
+        ttk.Button(left, text="Scan", command=self._scan).pack(fill=tk.X, pady=(0, 6))
+        ttk.Button(left, text="Connect", command=self._connect_selected).pack(fill=tk.X)
+        self.device_list = tk.Listbox(left, height=26)
+        self.device_list.pack(fill=tk.BOTH, expand=True, pady=8)
+        self.device_list.bind("<Double-Button-1>", lambda _event: self._connect_selected())
+        self.status = tk.StringVar(value="Idle")
+        ttk.Label(left, textvariable=self.status, wraplength=280).pack(fill=tk.X)
+
+        right = ttk.Frame(root)
+        root.add(right, weight=1)
+        tabs = ttk.Notebook(right)
+        tabs.pack(fill=tk.BOTH, expand=True)
+
+        self.details_tab = ttk.Frame(tabs)
+        self.settings_tab = ttk.Frame(tabs)
+        self.alarms_tab = ttk.Frame(tabs)
+        self.raw_tab = ttk.Frame(tabs)
+        tabs.add(self.details_tab, text="Details")
+        tabs.add(self.settings_tab, text="Settings")
+        tabs.add(self.alarms_tab, text="Alarms")
+        tabs.add(self.raw_tab, text="Raw")
+        self._build_details(self.details_tab)
+        self._build_settings(self.settings_tab)
+        self._build_alarms(self.alarms_tab)
+        self._build_raw(self.raw_tab)
+
+    def _value_row(self, parent: ttk.Frame, row: int, label: str, key: str) -> None:
+        ttk.Label(parent, text=label).grid(row=row, column=0, sticky="w", padx=8, pady=4)
+        var = self.vars.setdefault(key, tk.StringVar(value="--"))
+        ttk.Label(parent, textvariable=var, font=("TkDefaultFont", 10, "bold")).grid(row=row, column=1, sticky="w", padx=8, pady=4)
+
+    def _build_details(self, parent: ttk.Frame) -> None:
+        grid = ttk.Frame(parent)
+        grid.pack(anchor="nw", fill=tk.X, pady=8)
+        labels = [
+            ("Battery remaining", "remaining"), ("Discharge current", "current"), ("Discharge watts", "power"),
+            ("Voltage", "voltage"), ("Battery percent", "soc"), ("Time", "time"), ("Cell voltage max", "cell_max"),
+            ("Cell voltage min", "cell_min"), ("Cell imbalance", "cell_delta"), ("Cell count", "cell_count"),
+            ("Cycle count", "cycles"), ("Charge MOS", "charge_mos"), ("Discharge MOS", "discharge_mos"), ("State", "state"),
+        ]
+        for row, (label, key) in enumerate(labels):
+            self._value_row(grid, row, label, key)
+        self.cells_text = tk.Text(parent, height=8, wrap=tk.NONE)
+        self.cells_text.pack(fill=tk.BOTH, expand=True, padx=8, pady=8)
+
+    def _build_settings(self, parent: ttk.Frame) -> None:
+        top = ttk.Frame(parent)
+        top.pack(fill=tk.X, pady=8)
+        ttk.Label(top, text="PIN").pack(side=tk.LEFT, padx=8)
+        self.pin_entry = ttk.Entry(top, show="*", width=12)
+        self.pin_entry.pack(side=tk.LEFT)
+        ttk.Button(top, text="Unlock", command=self._unlock).pack(side=tk.LEFT, padx=8)
+        self.write_enabled = tk.BooleanVar(value=False)
+        ttk.Checkbutton(top, text="Enable MOS writes", variable=self.write_enabled).pack(side=tk.LEFT, padx=16)
+
+        info = ttk.Frame(parent)
+        info.pack(anchor="nw", fill=tk.X)
+        for row, (label, key) in enumerate([
+            ("Battery code", "battery_code"), ("Firmware", "firmware"), ("Production date", "production_date"),
+            ("Battery Ah", "rated_capacity"), ("Product info", "product_info"), ("Read PIN", "read_pin"),
+        ]):
+            self._value_row(info, row, label, key)
+
+        mos = ttk.LabelFrame(parent, text="MOS controls")
+        mos.pack(fill=tk.X, padx=8, pady=12)
+        ttk.Label(mos, text="Charge register").grid(row=0, column=0, padx=8, pady=6)
+        self.charge_reg = ttk.Entry(mos, width=8)
+        self.charge_reg.insert(0, "0x0035")
+        self.charge_reg.grid(row=0, column=1)
+        ttk.Button(mos, text="Charge ON", command=lambda: self._write_mos(self.charge_reg, True)).grid(row=0, column=2, padx=4)
+        ttk.Button(mos, text="Charge OFF", command=lambda: self._write_mos(self.charge_reg, False)).grid(row=0, column=3, padx=4)
+        ttk.Label(mos, text="Discharge register").grid(row=1, column=0, padx=8, pady=6)
+        self.discharge_reg = ttk.Entry(mos, width=8)
+        self.discharge_reg.insert(0, "0x0036")
+        self.discharge_reg.grid(row=1, column=1)
+        ttk.Button(mos, text="Discharge ON", command=lambda: self._write_mos(self.discharge_reg, True)).grid(row=1, column=2, padx=4)
+        ttk.Button(mos, text="Discharge OFF", command=lambda: self._write_mos(self.discharge_reg, False)).grid(row=1, column=3, padx=4)
+
+    def _build_alarms(self, parent: ttk.Frame) -> None:
+        self.alarm_text = tk.Text(parent, wrap=tk.WORD)
+        self.alarm_text.pack(fill=tk.BOTH, expand=True, padx=8, pady=8)
+
+    def _build_raw(self, parent: ttk.Frame) -> None:
+        self.raw_text = tk.Text(parent, wrap=tk.NONE)
+        self.raw_text.pack(fill=tk.BOTH, expand=True, padx=8, pady=8)
+
+    def _scan(self) -> None:
+        self.worker.scan()
+
+    def _connect_selected(self) -> None:
+        selection = self.device_list.curselection()
+        if not selection:
+            return
+        self.worker.connect(self.devices[selection[0]]["address"])
+
+    def _unlock(self) -> None:
+        entered = self.pin_entry.get().strip()
+        expected = self.vars.get("read_pin", tk.StringVar(value="000000")).get()
+        if entered and (expected in ("--", "") or entered == expected):
+            self.unlocked = True
+            self.status.set("Settings unlocked")
+        else:
+            messagebox.showerror("PIN", "PIN does not match the value read from the BMS")
+
+    def _write_mos(self, entry: ttk.Entry, enabled: bool) -> None:
+        if not self.unlocked or not self.write_enabled.get():
+            messagebox.showwarning("Writes disabled", "Enter the PIN and enable MOS writes first.")
+            return
+        try:
+            register = int(entry.get(), 0)
+        except ValueError:
+            messagebox.showerror("Register", "Register must be a number like 0x0035")
+            return
+        action = "ON" if enabled else "OFF"
+        if not messagebox.askyesno("Confirm write", f"Write {action} to register 0x{register:04X}?"):
+            return
+        self.worker.write_mos(register, enabled)
+
+    def _drain_queue(self) -> None:
+        while True:
+            try:
+                kind, payload = self.queue.get_nowait()
+            except queue.Empty:
+                break
+            if kind == "status":
+                self.status.set(str(payload))
+            elif kind == "error":
+                self.status.set(str(payload))
+                messagebox.showerror("BullTron BLE", str(payload))
+            elif kind == "scan":
+                self.devices = payload
+                self.device_list.delete(0, tk.END)
+                for row in self.devices:
+                    mark = "*" if row["likely"] else " "
+                    self.device_list.insert(tk.END, f"{mark} {row['name']}  {row['address']}  RSSI {row['rssi']}")
+            elif kind == "connected":
+                self.status.set(f"Connected: {payload}")
+            elif kind == "state":
+                self._render_state(payload)
+            elif kind in ("rx", "tx", "log"):
+                prefix = kind.upper()
+                self.raw_text.insert(tk.END, f"{prefix}: {payload}\n")
+                self.raw_text.see(tk.END)
+        self.after(100, self._drain_queue)
+
+    def _set(self, key: str, value: str) -> None:
+        self.vars.setdefault(key, tk.StringVar()).set(value)
+
+    def _render_state(self, state: DeviceState) -> None:
+        t = state.telemetry
+        self._set("remaining", f"{t.remaining_ah:.1f} Ah" if t.remaining_ah is not None else "--")
+        self._set("current", f"{t.current:.1f} A" if t.current is not None else "--")
+        self._set("power", f"{t.power_w:.0f} W" if t.power_w is not None else "--")
+        self._set("voltage", f"{t.voltage:.1f} V" if t.voltage is not None else "--")
+        self._set("soc", f"{t.soc:.1f} %" if t.soc is not None else "--")
+        self._set("time", f"{t.time_label}: {t.time_value}")
+        self._set("cell_max", f"{t.cell_max:.3f} V" if t.cell_max is not None else "--")
+        self._set("cell_min", f"{t.cell_min:.3f} V" if t.cell_min is not None else "--")
+        self._set("cell_delta", f"{t.cell_delta:.3f} V" if t.cell_delta is not None else "--")
+        self._set("cell_count", str(t.cell_count) if t.cell_count is not None else "--")
+        self._set("cycles", str(t.cycle_count) if t.cycle_count is not None else "--")
+        self._set("charge_mos", "ON" if t.charge_mos else "OFF" if t.charge_mos is not None else "--")
+        self._set("discharge_mos", "ON" if t.discharge_mos else "OFF" if t.discharge_mos is not None else "--")
+        self._set("state", {1: "charging", 2: "discharging"}.get(t.state, str(t.state) if t.state is not None else "--"))
+        self._set("battery_code", state.battery_code or "--")
+        self._set("firmware", state.firmware or "--")
+        self._set("production_date", state.production_date or "--")
+        self._set("rated_capacity", f"{state.rated_capacity_ah:.1f} Ah" if state.rated_capacity_ah is not None else "--")
+        self._set("product_info", state.product_info or "--")
+        self._set("read_pin", state.control_pin or "--")
+        self.cells_text.delete("1.0", tk.END)
+        if t.cells:
+            self.cells_text.insert(tk.END, "\n".join(f"Cell {i + 1:02d}: {v:.3f} V" for i, v in enumerate(t.cells)))
+        self.alarm_text.delete("1.0", tk.END)
+        if state.alarms:
+            self.alarm_text.insert(tk.END, "\n".join(state.alarms))
+        elif t.alarm_words:
+            self.alarm_text.insert(tk.END, "No decoded active alarms\n\nRaw words: " + " ".join(f"{w:04X}" for w in t.alarm_words))
+
+    def destroy(self) -> None:
+        self.worker.submit(self.worker.disconnect())
+        super().destroy()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="BullTron BLE desktop dashboard")
+    parser.parse_args()
+    if ttk is None or messagebox is None:
+        raise SystemExit(
+            "tkinter is not installed. On Debian/Ubuntu install it with: sudo apt install python3-tk"
+        )
+    app = BullTronGui()
+    app.mainloop()
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1 @@
+bleak>=0.22

tests/__init__.py

@@ -0,0 +1 @@
+

tests/test_bulltron_gui.py

@@ -0,0 +1,46 @@
+import unittest
+
+from bulltron_gui import DeviceState, decode_response, read_frame, split_frames
+
+
+class BullTronParserTest(unittest.TestCase):
+    def test_read_frame_crc(self):
+        self.assertEqual(read_frame(0, 62).hex().upper(), "D2030000003ED7B9")
+
+    def test_split_and_decode_live_frame(self):
+        words = [0] * 62
+        words[0:4] = [3306, 3312, 3308, 3311]
+        words[40] = 132
+        words[41] = 29925
+        words[42] = 652
+        words[43] = 3312
+        words[44] = 3306
+        words[47] = 2
+        words[48] = 1075
+        words[51] = 3
+        words[53] = 1
+        words[54] = 1
+        words[56] = 6
+        words[61] = 0x0010
+        payload = b"".join(word.to_bytes(2, "big") for word in words)
+        body = bytes([0xD2, 0x03, len(payload)]) + payload
+        from bulltron_gui import crc16_modbus_swapped
+
+        frame = body + crc16_modbus_swapped(body).to_bytes(2, "big")
+        buffer = bytearray(b"\x00" + frame + frame)
+        frames = split_frames(buffer)
+        self.assertEqual(len(frames), 2)
+
+        state = DeviceState()
+        decode_response(frames[0], state)
+        self.assertAlmostEqual(state.telemetry.voltage, 13.2)
+        self.assertAlmostEqual(state.telemetry.current, -7.5)
+        self.assertAlmostEqual(state.telemetry.soc, 65.2)
+        self.assertEqual(state.telemetry.cycle_count, 3)
+        self.assertTrue(state.telemetry.charge_mos)
+        self.assertTrue(state.telemetry.discharge_mos)
+        self.assertEqual(state.telemetry.cell_count, 4)
+
+
+if __name__ == "__main__":
+    unittest.main()