diff --git a/berger_gui.py b/berger_gui.py index 07bb429..58f714f 100644 --- a/berger_gui.py +++ b/berger_gui.py @@ -6,6 +6,7 @@ from __future__ import annotations import asyncio import threading from dataclasses import dataclass, field +from concurrent.futures import Future try: import tkinter as tk @@ -44,18 +45,28 @@ def write_frame(command: int, payload: bytes) -> bytes: return build_frame(0x5A, command, payload) +def mos_mode(charge_enabled: bool, discharge_enabled: bool) -> int: + return (0 if charge_enabled else 1) | (0 if discharge_enabled else 2) + + +def cell_statistics(cells: list[float]) -> tuple[float, float, float] | None: + if not cells: + return None + minimum = min(cells) + maximum = max(cells) + return minimum, maximum, round(maximum - minimum, 3) + + def validate_frame(frame: bytes) -> bool: if len(frame) < 7 or frame[0] != 0xDD or frame[-1] != 0x77: return False payload_len = frame[3] if len(frame) != payload_len + 7: return False - expected = checksum(frame[1], frame[2:4] + frame[4 : 4 + payload_len]) - # Responses include mode/status at [1]/[2]; requests use command at [2]. - response_total = (frame[1] + frame[2] + frame[3] + sum(frame[4 : 4 + payload_len])) & 0xFFFF - response_expected = ((response_total ^ 0xFFFF) + 1) & 0xFFFF + # Response checksums cover status, length, and payload, but not the command. + expected = (-sum(frame[2:-3])) & 0xFFFF actual = int.from_bytes(frame[-3:-1], "big") - return actual in {expected, response_expected} + return actual == expected def split_frames(buffer: bytearray) -> list[bytes]: @@ -196,14 +207,22 @@ class BergerApp: self.state = DeviceState() self.devices = [] self.client: BleakClient | None = None + self.write_response = False self.buffer = bytearray() + self.response_waiters: dict[int, asyncio.Future] = {} + self.poll_lock = asyncio.Lock() self.loop = asyncio.new_event_loop() threading.Thread(target=self.loop.run_forever, daemon=True).start() self.status = tk.StringVar(value="Idle") self.scan_seconds = tk.IntVar(value=6) - self.direct = tk.StringVar() + self.direct = tk.StringVar(value="") + self.charge_enabled = tk.BooleanVar(value=False) + self.discharge_enabled = tk.BooleanVar(value=False) + self.mos_ready = False + self.mos_busy = False self.rows: dict[str, tk.StringVar] = {} + self.cell_rows: list[tk.StringVar] = [] self._build_ui() def _build_ui(self) -> None: @@ -235,7 +254,10 @@ class BergerApp: ("cycle_count", "Cycles"), ("mos", "MOS"), ("temperatures", "Temps"), - ("cells", "Cells"), + ("cells", "Cell count"), + ("cell_min", "Cell min"), + ("cell_max", "Cell max"), + ("cell_delta", "Cell delta"), ("time", "Time"), ("production_date", "Production"), ("software_version", "Software"), @@ -246,13 +268,42 @@ class BergerApp: ttk.Label(data, text=label).grid(row=idx // 2, column=(idx % 2) * 2, sticky="e", padx=4, pady=2) ttk.Label(data, textvariable=self.rows[key], width=32).grid(row=idx // 2, column=(idx % 2) * 2 + 1, sticky="w") + cells = ttk.Frame(root) + cells.grid(row=3, column=0, sticky="ew", pady=(8, 2)) + ttk.Label(cells, text="Cell voltages").grid(row=0, column=0, sticky="ne", padx=4, pady=2) + self.cell_values = ttk.Frame(cells) + self.cell_values.grid(row=0, column=1, sticky="w") + controls = ttk.Frame(root) - controls.grid(row=3, column=0, sticky="ew", pady=8) + controls.grid(row=4, column=0, sticky="ew", pady=8) ttk.Button(controls, text="Refresh", command=self.poll_once).grid(row=0, column=0) ttk.Button(controls, text="Disconnect", command=self.disconnect).grid(row=0, column=1, padx=6) + self.charge_control = ttk.Checkbutton( + controls, + text="Charge enabled", + variable=self.charge_enabled, + command=self.set_mos, + ) + self.charge_control.grid(row=0, column=2, padx=(12, 6)) + self.discharge_control = ttk.Checkbutton( + controls, + text="Discharge enabled", + variable=self.discharge_enabled, + command=self.set_mos, + ) + self.discharge_control.grid(row=0, column=3, padx=6) + self._update_mos_controls() def run(self, coro): - return asyncio.run_coroutine_threadsafe(coro, self.loop) + future = asyncio.run_coroutine_threadsafe(coro, self.loop) + future.add_done_callback(self._handle_future) + return future + + def _handle_future(self, future: Future) -> None: + try: + future.result() + except Exception as exc: + self.root.after(0, lambda: self.status.set(f"BLE error: {exc}")) def scan(self) -> None: self.status.set("Scanning") @@ -286,6 +337,9 @@ class BergerApp: messagebox.showinfo("Berger LFP BLE", "Select a device or enter an address.") return address = self.devices[selected[0]][0].address + self.mos_ready = False + self.mos_busy = False + self._update_mos_controls() self.status.set("Connecting") self.run(self._connect(address)) @@ -295,24 +349,50 @@ class BergerApp: return self.client = BleakClient(address) await self.client.connect() + services = self.client.services + if services is None: + services = await self.client.get_services() + write_char = services.get_characteristic(WRITE_UUID) + notify_char = services.get_characteristic(NOTIFY_UUID) + if write_char is None: + raise RuntimeError(f"write characteristic not found: {WRITE_UUID}") + if notify_char is None: + raise RuntimeError(f"notify characteristic not found: {NOTIFY_UUID}") + props = set(write_char.properties) + self.write_response = "write-without-response" not in props and "write" in props await self.client.start_notify(NOTIFY_UUID, self._on_notify) - await self._send(read_frame(0x03)) - await self._send(read_frame(0x04)) - self.root.after(0, lambda: self.status.set("Connected")) + await self._poll_once() async def _send(self, data: bytes) -> None: if not self.client: return self.state.tx_log.append(data.hex().upper()) for off in range(0, len(data), 20): - await self.client.write_gatt_char(WRITE_UUID, data[off : off + 20], response=False) + await self.client.write_gatt_char(WRITE_UUID, data[off : off + 20], response=self.write_response) def _on_notify(self, _sender, data: bytearray) -> None: self.buffer.extend(data) + base_received = False for frame in split_frames(self.buffer): self.state.rx_log.append(frame.hex().upper()) decode_response(frame, self.state) - self.root.after(0, self._render_state) + waiter = self.response_waiters.get(frame[1]) + if waiter and not waiter.done(): + waiter.set_result(frame) + if validate_frame(frame) and frame[1] == 0x03 and frame[2] == 0: + base_received = True + self.root.after(0, lambda: self.status.set(f"Connected, TX {len(self.state.tx_log)}, RX {len(self.state.rx_log)}")) + self.root.after(0, self._base_received if base_received else self._render_state) + + def _base_received(self) -> None: + self.mos_ready = True + self._render_state() + self._update_mos_controls() + + def _update_mos_controls(self) -> None: + state = "normal" if self.mos_ready and not self.mos_busy else "disabled" + self.charge_control.configure(state=state) + self.discharge_control.configure(state=state) def _render_state(self) -> None: t = self.state.telemetry @@ -324,8 +404,15 @@ class BergerApp: self.rows["nominal_capacity"].set(f"{t.nominal_capacity:.2f} Ah") self.rows["cycle_count"].set(str(t.cycle_count)) self.rows["mos"].set(f"charge {'on' if t.charge_mos else 'off'}, discharge {'on' if t.discharge_mos else 'off'}") + self.charge_enabled.set(t.charge_mos) + self.discharge_enabled.set(t.discharge_mos) self.rows["temperatures"].set(", ".join(f"{v:.1f} C" for v in t.temperatures) or "--") - self.rows["cells"].set(f"{len(t.cells)} cells, {min(t.cells):.3f}-{max(t.cells):.3f} V" if t.cells else "--") + self.rows["cells"].set(str(len(t.cells)) if t.cells else "--") + stats = cell_statistics(t.cells) + self.rows["cell_min"].set(f"{stats[0]:.3f} V" if stats else "--") + self.rows["cell_max"].set(f"{stats[1]:.3f} V" if stats else "--") + self.rows["cell_delta"].set(f"{stats[2]:.3f} V" if stats else "--") + self._render_cells(t.cells) self.rows["time"].set(f"{t.time_label}: {t.time_value}") self.rows["production_date"].set(t.production_date or "--") self.rows["software_version"].set(t.software_version or "--") @@ -336,12 +423,102 @@ class BergerApp: status.append(f"protection bit {t.protection_bit}") self.rows["status"].set(", ".join(status) or "normal") + def _render_cells(self, cells: list[float]) -> None: + if len(self.cell_rows) != len(cells): + for widget in self.cell_values.winfo_children(): + widget.destroy() + self.cell_rows = [] + for index in range(len(cells)): + row = index // 4 + column = (index % 4) * 2 + value = tk.StringVar() + self.cell_rows.append(value) + ttk.Label(self.cell_values, text=f"Cell {index + 1}").grid( + row=row, column=column, sticky="e", padx=(4, 2), pady=2 + ) + ttk.Label(self.cell_values, textvariable=value, width=9).grid( + row=row, column=column + 1, sticky="w", padx=(0, 8), pady=2 + ) + for value, voltage in zip(self.cell_rows, cells): + value.set(f"{voltage:.3f} V") + def poll_once(self) -> None: self.run(self._poll_once()) + async def _exchange(self, command: int, frame: bytes) -> bytes | None: + waiter = self.loop.create_future() + self.response_waiters[command] = waiter + try: + await self._send(frame) + return await asyncio.wait_for(waiter, timeout=2.0) + except TimeoutError: + return None + finally: + if self.response_waiters.get(command) is waiter: + del self.response_waiters[command] + + async def _request(self, command: int) -> bool: + response = await self._exchange(command, read_frame(command)) + return bool(response and validate_frame(response) and response[2] == 0) + async def _poll_once(self) -> None: - await self._send(read_frame(0x03)) - await self._send(read_frame(0x04)) + if not self.client or not self.client.is_connected: + self.root.after(0, lambda: self.status.set("Not connected")) + return + async with self.poll_lock: + missing = [] + for index, command in enumerate((0x03, 0x04)): + if index: + await asyncio.sleep(0.15) + if not await self._request(command): + missing.append(f"0x{command:02X}") + suffix = f", no response from {', '.join(missing)}" if missing else "" + self.root.after( + 0, + lambda: self.status.set( + f"Connected, TX {len(self.state.tx_log)}, RX {len(self.state.rx_log)}{suffix}" + ), + ) + + def set_mos(self) -> None: + if not self.mos_ready or self.mos_busy: + self._render_state() + return + charge_enabled = self.charge_enabled.get() + discharge_enabled = self.discharge_enabled.get() + self.mos_busy = True + self._update_mos_controls() + self.status.set("Updating MOS state") + self.run(self._set_mos(charge_enabled, discharge_enabled)) + + async def _set_mos(self, charge_enabled: bool, discharge_enabled: bool) -> None: + if not self.client or not self.client.is_connected: + self.root.after(0, lambda: self._finish_mos_change("Not connected", ready=False)) + return + async with self.poll_lock: + mode = mos_mode(charge_enabled, discharge_enabled) + frame = write_frame(0xE1, bytes([0x00, mode])) + response = await self._exchange(0xE1, frame) + accepted = bool(response and validate_frame(response) and response[2] == 0) + refreshed = False + if accepted: + await asyncio.sleep(0.15) + refreshed = await self._request(0x03) + + if not accepted: + message = "MOS command was not acknowledged" + elif not refreshed: + message = "MOS updated; telemetry refresh timed out" + else: + message = f"MOS updated, TX {len(self.state.tx_log)}, RX {len(self.state.rx_log)}" + self.root.after(0, lambda: self._finish_mos_change(message, ready=True)) + + def _finish_mos_change(self, message: str, ready: bool) -> None: + self.mos_busy = False + self.mos_ready = ready + self._render_state() + self._update_mos_controls() + self.status.set(message) def disconnect(self) -> None: self.run(self._disconnect()) @@ -350,7 +527,13 @@ class BergerApp: if self.client: await self.client.disconnect() self.client = None - self.root.after(0, lambda: self.status.set("Disconnected")) + self.root.after(0, self._show_disconnected) + + def _show_disconnected(self) -> None: + self.mos_ready = False + self.mos_busy = False + self._update_mos_controls() + self.status.set("Disconnected") def main() -> None: diff --git a/tests/test_berger_gui.py b/tests/test_berger_gui.py index a0a8da2..0dfe7de 100644 --- a/tests/test_berger_gui.py +++ b/tests/test_berger_gui.py @@ -1,6 +1,22 @@ import unittest -from berger_gui import DeviceState, berger_match_reason, decode_response, read_frame, split_frames, write_frame +from berger_gui import ( + DeviceState, + berger_match_reason, + cell_statistics, + decode_response, + mos_mode, + read_frame, + split_frames, + validate_frame, + write_frame, +) + + +def response_frame(command, payload): + body = bytes([0xDD, command, 0x00, len(payload), *payload]) + check = (-sum(body[2:])) & 0xFFFF + return body + check.to_bytes(2, "big") + b"\x77" class BergerParserTest(unittest.TestCase): @@ -11,6 +27,12 @@ class BergerParserTest(unittest.TestCase): def test_write_switch_checksum(self): self.assertEqual(write_frame(0xE1, b"\x00\x00").hex().upper(), "DD5AE1020000FF1D77") + def test_mos_modes(self): + self.assertEqual(mos_mode(True, True), 0) + self.assertEqual(mos_mode(False, True), 1) + self.assertEqual(mos_mode(True, False), 2) + self.assertEqual(mos_mode(False, False), 3) + def test_decode_base_data(self): payload = bytearray(27 + 4) payload[0:2] = (1328).to_bytes(2, "big") @@ -26,10 +48,7 @@ class BergerParserTest(unittest.TestCase): payload[22] = 2 payload[23:25] = (2981).to_bytes(2, "big") payload[25:27] = (2991).to_bytes(2, "big") - body = bytes([0xDD, 0x03, 0x00, len(payload), *payload]) - total = sum(body[1:]) & 0xFFFF - check = ((total ^ 0xFFFF) + 1) & 0xFFFF - frame = body + check.to_bytes(2, "big") + b"\x77" + frame = response_frame(0x03, payload) buffer = bytearray(b"\x00" + frame + frame) frames = split_frames(buffer) @@ -52,14 +71,47 @@ class BergerParserTest(unittest.TestCase): def test_decode_cell_voltages(self): payload = b"".join(v.to_bytes(2, "big") for v in [3306, 3312, 3308, 3311]) - body = bytes([0xDD, 0x04, 0x00, len(payload), *payload]) - total = sum(body[1:]) & 0xFFFF - check = ((total ^ 0xFFFF) + 1) & 0xFFFF - frame = body + check.to_bytes(2, "big") + b"\x77" + frame = response_frame(0x04, payload) state = DeviceState() decode_response(frame, state) self.assertEqual(state.telemetry.cells, [3.306, 3.312, 3.308, 3.311]) + def test_captured_fragmented_base_data(self): + chunks = [ + bytes.fromhex("DD030026053400005F8C75300000348900000000"), + bytes.fromhex("000063520304030BB70BA60BAA00000075305F8C"), + bytes.fromhex("0000F8DD77"), + ] + buffer = bytearray() + frames = [] + for chunk in chunks: + buffer.extend(chunk) + frames.extend(split_frames(buffer)) + + self.assertEqual(len(frames), 1) + self.assertTrue(validate_frame(frames[0])) + state = DeviceState() + decode_response(frames[0], state) + self.assertEqual(state.telemetry.voltage, 13.32) + self.assertEqual(state.telemetry.remaining_capacity, 244.6) + self.assertEqual(state.telemetry.nominal_capacity, 300.0) + self.assertEqual(state.telemetry.soc, 82) + self.assertEqual(state.telemetry.production_date, "2026-4-9") + self.assertEqual(state.telemetry.temperatures, [26.8, 25.1, 25.5]) + self.assertTrue(state.telemetry.charge_mos) + self.assertTrue(state.telemetry.discharge_mos) + + def test_captured_cell_voltages(self): + frame = bytes.fromhex("DD0400080D030D040D040D04FFB577") + self.assertTrue(validate_frame(frame)) + state = DeviceState() + decode_response(frame, state) + self.assertEqual(state.telemetry.cells, [3.331, 3.332, 3.332, 3.332]) + self.assertEqual(cell_statistics(state.telemetry.cells), (3.331, 3.332, 0.001)) + + def test_empty_cell_statistics(self): + self.assertIsNone(cell_statistics([])) + def test_scan_filter(self): self.assertEqual(berger_match_reason(None, ["0000ff00-0000-1000-8000-00805f9b34fb"]), "service:ff00") self.assertEqual(berger_match_reason("JBD-SP04S", []), "name")