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

Fix BullTron MOS refresh ordering

Browse Source
This commit is contained in:
acidburns
2026-07-11 15:47:10 +02:00
parent 9c3757f91d
commit 421cea6ea1
2 changed files with 66 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bulltron_gui.py
+43 -19
View File
@@ -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
tests/test_bulltron_gui.py
+23
View File
@@ -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")