C3MA/bulltron-ble-commands
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix BullTron GUI telemetry and MOS writes

Browse Source
This commit is contained in:
acidburns
2026-07-11 12:01:16 +02:00
parent b45641b2aa
commit 2addde3dbc
5 changed files with 116 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+5 -5
View File
@@ -146,8 +146,8 @@ at least one advertisement from that peripheral in the current session.
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
imbalance, cell count, cycle count, charge/discharge MOS, separate
charge/discharge current and watts, 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
@@ -156,9 +156,9 @@ The GUI shows:
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.
The default write registers mirror the BullTron system screen: `0x00A5` for
charge MOS and `0x00A6` for discharge MOS. The live MOS status still comes from
read-only telemetry registers `0x0035` and `0x0036`.
Scan results are sorted so likely BullTron devices appear first. The Android app
accepts scanned devices whose BLE name contains `DL` or `B35`, or whose legacy
bulltron_gui.py
+63 -21
View File
@@ -144,6 +144,10 @@ def format_duration(minutes: float | None) -> str:
return f"{hours} h {mins:02d} m"
def clean_ascii(data: bytes) -> str:
return printable_ascii(data).replace("\x00", "").strip()
def split_frames(buffer: bytearray) -> list[bytes]:
frames: list[bytes] = []
while True:
@@ -182,6 +186,10 @@ class Telemetry:
soc: float | None = None
remaining_ah: float | None = None
power_w: float | None = None
charge_current: float | None = None
discharge_current: float | None = None
charge_power_w: float | None = None
discharge_power_w: float | None = None
state: int | None = None
cycle_count: int | None = None
charge_mos: bool | None = None
@@ -245,16 +253,21 @@ def decode_live(payload: bytes, state: DeviceState) -> None:
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)
current_abs = abs(tel.current)
tel.power_w = current_abs * tel.voltage
tel.charge_current = current_abs if tel.state == 1 else 0.0 if tel.state == 2 else None
tel.discharge_current = current_abs if tel.state == 2 else 0.0 if tel.state == 1 else None
tel.charge_power_w = tel.voltage * tel.charge_current if tel.charge_current is not None else None
tel.discharge_power_w = tel.voltage * tel.discharge_current if tel.discharge_current is not None else None
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
minutes = capacity * ((100 - tel.soc) / 100) / abs(tel.current) * 60
elif tel.state == 2:
tel.time_label = "Time till empty"
minutes = capacity * tel.soc / abs(tel.current) * 60
minutes = capacity * (tel.soc / 100) / abs(tel.current) * 60
else:
tel.time_label = "Time"
tel.time_value = format_duration(minutes)
@@ -282,29 +295,52 @@ def decode_response(frame: bytes, state: DeviceState) -> None:
if byte_count == 0x7C:
decode_live(payload, state)
elif byte_count == 0x06:
text = printable_ascii(payload)
text = clean_ascii(payload)
state.control_pin = text or payload.hex().upper()
elif byte_count == 0x40:
state.firmware = printable_ascii(payload) or payload.hex().upper()
parse_version_block(payload, state)
elif byte_count == 0x18:
state.battery_code = printable_ascii(payload) or payload.hex().upper()
state.battery_code = clean_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}"
state.production_date = parse_production_date(payload)
elif byte_count == 0x4A:
parse_device_parameter_block(payload, state)
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)
if words:
state.rated_capacity_ah = words[0] * 0.1
def parse_version_block(payload: bytes, state: DeviceState) -> None:
app = clean_ascii(payload[0:16])
mcu = clean_ascii(payload[16:32])
machine = clean_ascii(payload[32:64])
if app or mcu:
state.firmware = " / ".join(part for part in (app[::-1] if app else "", mcu) if part)
if machine:
state.battery_code = machine
def parse_production_date(payload: bytes) -> str:
if len(payload) < 3 or payload[:3] in (b"\xff\xff\xff", b"\x00\x00\x00"):
return "2000-01-01"
year, month, day = payload[0], payload[1], payload[2]
if not 1 <= month <= 12 or not 1 <= day <= 31:
return payload.hex(" ").upper()
return f"20{year:02d}-{month:02d}-{day:02d}"
def parse_device_parameter_block(payload: bytes, state: DeviceState) -> None:
if len(payload) >= 64:
parse_version_block(payload[:64], state)
if len(payload) >= 70:
state.control_pin = clean_ascii(payload[64:70]) or payload[64:70].hex().upper()
if len(payload) >= 74:
state.production_date = parse_production_date(payload[70:74])
class BleWorker:
@@ -500,7 +536,8 @@ class BullTronGui(tk.Tk):
grid = ttk.Frame(parent)
grid.pack(anchor="nw", fill=tk.X, pady=8)
labels = [
("Battery remaining", "remaining"), ("Discharge current", "current"), ("Discharge watts", "power"),
("Battery remaining", "remaining"), ("Charge current", "charge_current"), ("Charge watts", "charge_power"),
("Discharge current", "discharge_current"), ("Discharge watts", "discharge_power"), ("Signed current", "current"),
("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"),
@@ -532,13 +569,13 @@ class BullTronGui(tk.Tk):
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.insert(0, "0x00A5")
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.insert(0, "0x00A6")
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)
@@ -575,7 +612,8 @@ class BullTronGui(tk.Tk):
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):
normalized_expected = "".join(ch for ch in expected if ch.isdigit())
if entered and (expected in ("--", "") or not normalized_expected or entered == normalized_expected):
self.unlocked = True
self.status.set("Settings unlocked")
else:
@@ -588,7 +626,7 @@ class BullTronGui(tk.Tk):
try:
register = int(entry.get(), 0)
except ValueError:
messagebox.showerror("Register", "Register must be a number like 0x0035")
messagebox.showerror("Register", "Register must be a number like 0x00A5")
return
action = "ON" if enabled else "OFF"
if not messagebox.askyesno("Confirm write", f"Write {action} to register 0x{register:04X}?"):
@@ -631,6 +669,10 @@ class BullTronGui(tk.Tk):
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("charge_current", f"{t.charge_current:.1f} A" if t.charge_current is not None else "--")
self._set("discharge_current", f"{t.discharge_current:.1f} A" if t.discharge_current is not None else "--")
self._set("charge_power", f"{t.charge_power_w:.0f} W" if t.charge_power_w is not None else "--")
self._set("discharge_power", f"{t.discharge_power_w:.0f} W" if t.discharge_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}")
docs/ble-flow.md
+2 -2
View File
@@ -95,10 +95,10 @@ Time-to-empty/full is not a separate BLE value in the live view. The app compute
```text
if register 0x002F == 1:
time-to-full = (rated_capacity_Ah * (100 - SOC_percent) / abs(current_A)) * 60 minutes
time-to-full = (rated_capacity_Ah * ((100 - SOC_percent) / 100) / abs(current_A)) * 60 minutes
if register 0x002F == 2:
time-to-empty = (rated_capacity_Ah * SOC_percent / abs(current_A)) * 60 minutes
time-to-empty = (rated_capacity_Ah * (SOC_percent / 100) / abs(current_A)) * 60 minutes
```
The `rated_capacity_Ah` input comes from the settings register set parsed into `contentByteArrayByReadSet`. The app also has a separate read of register `0x0064` length `1` that stores `residueTime`, but the visible live time calculation above is derived from SOC/current/capacity unless current is zero.
docs/command-catalog.md
+7 -2
View File
@@ -182,11 +182,11 @@ rated_capacity_Ah = parsed settings capacity
if register 0x002F == 1:
label = TimeToFull
minutes = rated_capacity_Ah * (100 - soc_percent) / current_A * 60
minutes = rated_capacity_Ah * ((100 - soc_percent) / 100) / current_A * 60
if register 0x002F == 2:
label = TimetoEmpty
minutes = rated_capacity_Ah * soc_percent / current_A * 60
minutes = rated_capacity_Ah * (soc_percent / 100) / current_A * 60
```
If current is zero or the state is neither `1` nor `2`, the UI shows `-- h --m`.
@@ -203,6 +203,11 @@ If current is zero or the state is neither `1` nor `2`, the UI shows `-- h --m`.
| `40` | version data |
| `06` | password data |
| `12` | balancing state |
In the BullTron system screen, rated capacity is settings word `0` scaled by
`0.1 Ah`. The charge and discharge MOS controls write single registers
`0x00A5` and `0x00A6`; live MOS status is read separately from telemetry
registers `0x0035` and `0x0036`.
| `0A` | new alarm info |
| `4A` | device parameter info |
| `04` | communication/protocol or production date |
tests/test_bulltron_gui.py
+39 -1
View File
@@ -1,6 +1,6 @@
import unittest
from bulltron_gui import DeviceState, bulltron_match_reason, decode_response, read_frame, split_frames
from bulltron_gui import DeviceState, bulltron_match_reason, decode_response, read_frame, split_frames, write_single_frame
class BullTronParserTest(unittest.TestCase):
@@ -35,11 +35,49 @@ class BullTronParserTest(unittest.TestCase):
decode_response(frames[0], state)
self.assertAlmostEqual(state.telemetry.voltage, 13.2)
self.assertAlmostEqual(state.telemetry.current, -7.5)
self.assertAlmostEqual(state.telemetry.discharge_current, 7.5)
self.assertAlmostEqual(state.telemetry.charge_current, 0.0)
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)
self.assertEqual(state.telemetry.time_value, "9 h 21 m")
def test_charging_time_uses_soc_fraction_and_rated_capacity(self):
state = DeviceState(rated_capacity_ah=160)
words = [0] * 62
words[40] = 132
words[41] = 30152
words[42] = 450
words[47] = 1
words[48] = 72
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")
decode_response(frame, state)
self.assertAlmostEqual(state.telemetry.charge_current, 15.2)
self.assertAlmostEqual(state.telemetry.charge_power_w, 200.64)
self.assertAlmostEqual(state.telemetry.discharge_current, 0.0)
self.assertEqual(state.telemetry.time_label, "Time till full")
self.assertEqual(state.telemetry.time_value, "5 h 47 m")
def test_settings_capacity_uses_app_word_zero(self):
state = DeviceState()
words = [1600, 3750, 42] + [0] * 38
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")
decode_response(frame, state)
self.assertAlmostEqual(state.rated_capacity_ah, 160.0)
def test_system_mos_write_registers(self):
self.assertEqual(write_single_frame(0x00A5, 1).hex().upper()[:8], "D20600A5")
self.assertEqual(write_single_frame(0x00A6, 0).hex().upper()[:8], "D20600A6")
def test_bulltron_scan_filter_matches_apk_markers(self):
self.assertEqual(bulltron_match_reason("DL-1234", [], []), "name:DL")