diff --git a/README.md b/README.md index 6ecd803..425f642 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/bulltron_gui.py b/bulltron_gui.py index 723e595..fbbecba 100644 --- a/bulltron_gui.py +++ b/bulltron_gui.py @@ -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}") diff --git a/docs/ble-flow.md b/docs/ble-flow.md index ad0ce81..91d0536 100644 --- a/docs/ble-flow.md +++ b/docs/ble-flow.md @@ -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. diff --git a/docs/command-catalog.md b/docs/command-catalog.md index 3850129..5d43aa3 100644 --- a/docs/command-catalog.md +++ b/docs/command-catalog.md @@ -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 | diff --git a/tests/test_bulltron_gui.py b/tests/test_bulltron_gui.py index c284cab..3a707a3 100644 --- a/tests/test_bulltron_gui.py +++ b/tests/test_bulltron_gui.py @@ -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")