From 421cea6ea1164e0b2fb9e4ec8daadd5b4a48b0d2 Mon Sep 17 00:00:00 2001 From: acidburns Date: Sat, 11 Jul 2026 15:47:10 +0200 Subject: [PATCH] Fix BullTron MOS refresh ordering --- bulltron_gui.py | 62 ++++++++++++++++++++++++++------------ tests/test_bulltron_gui.py | 23 ++++++++++++++ 2 files changed, 66 insertions(+), 19 deletions(-) diff --git a/bulltron_gui.py b/bulltron_gui.py index 645d7e4..b555ba9 100644 --- a/bulltron_gui.py +++ b/bulltron_gui.py @@ -300,7 +300,7 @@ def decode_alarm_words(words: list[int]) -> list[str]: return active -def decode_response(frame: bytes, state: DeviceState) -> None: +def decode_response(frame: bytes, state: DeviceState, read_start: int | None = None) -> None: if frame[1] != 0x03: state.raw_blocks["Last ACK"] = frame.hex(" ").upper() return @@ -318,21 +318,30 @@ def decode_response(frame: bytes, state: DeviceState) -> None: parse_version_block(payload, state) elif byte_count == 0x18: state.battery_code = clean_ascii(payload) or payload.hex().upper() + elif byte_count == 0x04 and read_start == 0x00A5: + words = words_from_payload(payload) + if len(words) >= 2: + state.control_charge_mos = words[0] != 0 + state.control_discharge_mos = words[1] != 0 elif byte_count == 0x04: 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) + parse_settings(payload, state, read_start) -def parse_settings(payload: bytes, state: DeviceState) -> None: +def parse_settings(payload: bytes, state: DeviceState, read_start: int | None = 0x0080) -> 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 + start = 0x0080 if read_start is None else read_start + charge_index = 0x00A5 - start + discharge_index = 0x00A6 - start + if 0 <= charge_index < len(words): + state.control_charge_mos = words[charge_index] != 0 + if 0 <= discharge_index < len(words): + state.control_discharge_mos = words[discharge_index] != 0 def parse_version_block(payload: bytes, state: DeviceState) -> None: @@ -371,6 +380,8 @@ class BleWorker: self.thread.start() self.client: Any = None self.buffer = bytearray() + self.pending_reads: list[tuple[int, int]] = [] + self.command_lock = asyncio.Lock() self.state = DeviceState() self.polling = False @@ -441,6 +452,8 @@ class BleWorker: async def _write_command(self, data: bytes, response: bool = False) -> None: if self.client and self.client.is_connected: + if len(data) >= 6 and data[0] == 0xD2 and data[1] == 0x03: + self.pending_reads.append((int.from_bytes(data[2:4], "big"), int.from_bytes(data[4:6], "big") * 2)) await self.client.write_gatt_char(WRITE_CHAR, data, response=response) self.ui_queue.put(("tx", data.hex(" ").upper())) @@ -448,12 +461,13 @@ class BleWorker: 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)) + async with self.command_lock: + 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}")) @@ -463,7 +477,14 @@ class BleWorker: 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) + read_start = None + if frame[1] == 0x03 and self.pending_reads: + byte_count = frame[2] + match_index = next((i for i, (_, expected_bytes) in enumerate(self.pending_reads) if expected_bytes == byte_count), 0) + read_start, _ = self.pending_reads.pop(match_index) + if match_index: + del self.pending_reads[:match_index] + decode_response(frame, self.state, read_start) self.ui_queue.put(("rx", frame.hex(" ").upper())) self.ui_queue.put(("state", self.state)) @@ -484,12 +505,15 @@ class BleWorker: 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 with self.command_lock: + 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(0x00A5, 0x0002)) + await asyncio.sleep(0.12) + 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 acknowledged; refreshed 0x00A5/0x00A6 and live MOS status")) async def disconnect(self) -> None: self.polling = False diff --git a/tests/test_bulltron_gui.py b/tests/test_bulltron_gui.py index f920bf0..030a367 100644 --- a/tests/test_bulltron_gui.py +++ b/tests/test_bulltron_gui.py @@ -121,6 +121,29 @@ class BullTronParserTest(unittest.TestCase): self.assertTrue(state.control_charge_mos) self.assertFalse(state.control_discharge_mos) + def test_direct_mos_control_read_uses_register_context(self): + state = DeviceState() + payload = b"\x00\x00\x00\x01" + 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, 0x00A5) + + self.assertFalse(state.control_charge_mos) + self.assertTrue(state.control_discharge_mos) + + def test_four_byte_non_mos_read_still_decodes_production_date(self): + state = DeviceState() + payload = b"\x19\x01\x01\x00" + 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, 0x00C9) + + self.assertEqual(state.production_date, "2025-01-01") + 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")