655 lines
27 KiB
Python
655 lines
27 KiB
Python
#!/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"
|
|
BULLTRON_NAME_MARKERS = ("dl", "b35")
|
|
BULLTRON_ADV_MARKERS = (b"DL", b"PU", b"JHB")
|
|
|
|
|
|
def iter_advertisement_payloads(adv: Any) -> list[bytes]:
|
|
"""Best-effort extraction of payload-like byte blobs from bleak metadata."""
|
|
payloads: list[bytes] = []
|
|
for data in getattr(adv, "manufacturer_data", {}).values():
|
|
if isinstance(data, bytes):
|
|
payloads.append(data)
|
|
for data in getattr(adv, "service_data", {}).values():
|
|
if isinstance(data, bytes):
|
|
payloads.append(data)
|
|
for item in getattr(adv, "platform_data", ()) or ():
|
|
if isinstance(item, bytes):
|
|
payloads.append(item)
|
|
elif isinstance(item, (tuple, list)):
|
|
payloads.extend(part for part in item if isinstance(part, bytes))
|
|
return payloads
|
|
|
|
|
|
def bulltron_match_reason(name: str, service_uuids: list[str], payloads: list[bytes] | None = None) -> str | None:
|
|
lowered = name.lower()
|
|
if "bull" in lowered:
|
|
return "name:BullTron"
|
|
for marker in BULLTRON_NAME_MARKERS:
|
|
if marker in lowered:
|
|
return f"name:{marker.upper()}"
|
|
if MAIN_SERVICE in service_uuids:
|
|
return "service:fff0"
|
|
for payload in payloads or []:
|
|
# The Android app checks the legacy scanRecord at byte offsets 14..16
|
|
# for ASCII DL/PU/JHB markers, then requires byte 16 to be 0x00.
|
|
if len(payload) > 17 and payload[16] == 0x00:
|
|
for marker in BULLTRON_ADV_MARKERS:
|
|
if payload[14 : 14 + len(marker)].upper() == marker:
|
|
return f"adv:{marker.decode('ascii')}"
|
|
for marker in BULLTRON_ADV_MARKERS:
|
|
if marker in payload.upper():
|
|
return f"adv:{marker.decode('ascii')}"
|
|
return None
|
|
|
|
|
|
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 [])]
|
|
reason = bulltron_match_reason(name, service_uuids, iter_advertisement_payloads(adv))
|
|
rows.append({"name": name, "address": dev.address, "rssi": adv.rssi, "likely": bool(reason), "reason": reason or ""})
|
|
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 " "
|
|
reason = f" [{row['reason']}]" if row["reason"] else ""
|
|
self.device_list.insert(tk.END, f"{mark} {row['name']} {row['address']} RSSI {row['rssi']}{reason}")
|
|
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()
|