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

Fix BullTron empty time and MOS refresh

Browse Source
This commit is contained in:
acidburns
2026-07-11 12:46:07 +02:00
parent bff0af18c0
commit 9c3757f91d
2 changed files with 27 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bulltron_gui.py
+20 -17
View File
@@ -208,6 +208,8 @@ class Telemetry:
class DeviceState:
telemetry: Telemetry = field(default_factory=Telemetry)
rated_capacity_ah: float | None = None
control_charge_mos: bool | None = None
control_discharge_mos: bool | None = None
control_pin: str | None = None
firmware: str = ""
battery_code: str = ""
@@ -262,20 +264,6 @@ def decode_live(payload: bytes, state: DeviceState) -> None:
tel.discharge_current = current_abs if is_discharging else 0.0 if is_charging 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
if tel.residue_minutes is not None and tel.residue_minutes != 0xFFFF:
if tel.current is not None and tel.current < -0.05:
tel.time_label = "Time till empty"
elif tel.current is not None and tel.current > 0.05:
tel.time_label = "Time till full"
elif tel.state == 2:
tel.time_label = "Time till empty"
elif tel.state == 1:
tel.time_label = "Time till full"
else:
tel.time_label = "Time"
tel.time_value = format_duration(tel.residue_minutes)
state.alarms = decode_alarm_words(tel.alarm_words)
return
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:
@@ -342,6 +330,9 @@ def parse_settings(payload: bytes, state: DeviceState) -> None:
words = words_from_payload(payload)
if words:
state.rated_capacity_ah = words[0] * 0.1
if len(words) > 38:
state.control_charge_mos = words[0x00A5 - 0x0080] != 0
state.control_discharge_mos = words[0x00A6 - 0x0080] != 0
def parse_version_block(payload: bytes, state: DeviceState) -> None:
@@ -448,9 +439,9 @@ class BleWorker:
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:
async def _write_command(self, data: bytes, response: bool = False) -> None:
if self.client and self.client.is_connected:
await self.client.write_gatt_char(WRITE_CHAR, data, response=False)
await self.client.write_gatt_char(WRITE_CHAR, data, response=response)
self.ui_queue.put(("tx", data.hex(" ").upper()))
async def _poll_loop(self) -> None:
@@ -489,7 +480,16 @@ class BleWorker:
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)))
self.submit(self._write_mos(register, enabled))
async def _write_mos(self, register: int, enabled: bool) -> None:
self.ui_queue.put(("status", f"Writing MOS register 0x{register:04X}..."))
await self._write_command(write_single_frame(register, 1 if enabled else 0), response=True)
await asyncio.sleep(0.5)
await self._write_command(read_frame(0x0000, 0x003E))
await asyncio.sleep(0.12)
await self._write_command(read_frame(0x0080, 0x0029))
self.ui_queue.put(("status", f"MOS write sent; refreshed 0x0035/0x0036 and 0x00A5/0x00A6"))
async def disconnect(self) -> None:
self.polling = False
@@ -592,6 +592,7 @@ class BullTronGui(tk.Tk):
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"),
("Charge MOS control", "control_charge_mos"), ("Discharge MOS control", "control_discharge_mos"),
]):
self._value_row(info, row, label, key)
@@ -724,6 +725,8 @@ class BullTronGui(tk.Tk):
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._set("control_charge_mos", "ON" if state.control_charge_mos else "OFF" if state.control_charge_mos is not None else "--")
self._set("control_discharge_mos", "ON" if state.control_discharge_mos else "OFF" if state.control_discharge_mos is not None else "--")
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)))
tests/test_bulltron_gui.py
+7 -5
View File
@@ -85,9 +85,9 @@ class BullTronParserTest(unittest.TestCase):
self.assertEqual(state.telemetry.time_label, "Time till empty")
self.assertEqual(state.telemetry.time_value, "10 h 03 m")
def test_bms_residue_time_register_overrides_estimate(self):
def test_residue_register_does_not_hide_estimated_empty_time(self):
state = DeviceState(rated_capacity_ah=160)
residue_payload = (6 * 60 + 30).to_bytes(2, "big")
residue_payload = (0).to_bytes(2, "big")
residue_body = bytes([0xD2, 0x03, len(residue_payload)]) + residue_payload
from bulltron_gui import crc16_modbus_swapped
@@ -96,7 +96,7 @@ class BullTronParserTest(unittest.TestCase):
words = [0] * 62
words[40] = 128
words[41] = 29999
words[41] = 29922
words[42] = 490
words[47] = 2
words[48] = 784
@@ -106,11 +106,11 @@ class BullTronParserTest(unittest.TestCase):
decode_response(live_frame, state)
self.assertEqual(state.telemetry.time_label, "Time till empty")
self.assertEqual(state.telemetry.time_value, "6 h 30 m")
self.assertEqual(state.telemetry.time_value, "10 h 03 m")
def test_settings_capacity_uses_app_word_zero(self):
state = DeviceState()
words = [1600, 3750, 42] + [0] * 38
words = [1600, 3750, 42] + [0] * 34 + [1, 0, 99, 0]
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
@@ -118,6 +118,8 @@ class BullTronParserTest(unittest.TestCase):
frame = body + crc16_modbus_swapped(body).to_bytes(2, "big")
decode_response(frame, state)
self.assertAlmostEqual(state.rated_capacity_ah, 160.0)
self.assertTrue(state.control_charge_mos)
self.assertFalse(state.control_discharge_mos)
def test_system_mos_write_registers(self):
self.assertEqual(write_single_frame(0x00A5, 1).hex().upper()[:8], "D20600A5")