From 9c3757f91d340c171f83ffafc433815fd62055af Mon Sep 17 00:00:00 2001 From: acidburns Date: Sat, 11 Jul 2026 12:46:07 +0200 Subject: [PATCH] Fix BullTron empty time and MOS refresh --- bulltron_gui.py | 37 ++++++++++++++++++++----------------- tests/test_bulltron_gui.py | 12 +++++++----- 2 files changed, 27 insertions(+), 22 deletions(-) diff --git a/bulltron_gui.py b/bulltron_gui.py index 70c74de..645d7e4 100644 --- a/bulltron_gui.py +++ b/bulltron_gui.py @@ -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))) diff --git a/tests/test_bulltron_gui.py b/tests/test_bulltron_gui.py index 602e5af..f920bf0 100644 --- a/tests/test_bulltron_gui.py +++ b/tests/test_bulltron_gui.py @@ -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")