diff --git a/.gitignore b/.gitignore index 4bb7642..1cade5a 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,6 @@ apk/ *.dex *.jar *.class +__pycache__/ +*.py[cod] +.venv/ diff --git a/README.md b/README.md index 29238bf..c8fade1 100644 --- a/README.md +++ b/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: diff --git a/bulltron_gui.py b/bulltron_gui.py new file mode 100644 index 0000000..6be51a2 --- /dev/null +++ b/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("", 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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..af8eee2 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +bleak>=0.22 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/test_bulltron_gui.py b/tests/test_bulltron_gui.py new file mode 100644 index 0000000..a359207 --- /dev/null +++ b/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()