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

Improve Berger BLE telemetry and controls

Browse Source
This commit is contained in:
acidburns
2026-07-31 21:03:29 +02:00
parent b390792e34
commit 6cd6a31435
2 changed files with 262 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
berger_gui.py
+201 -18
View File
@@ -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:
tests/test_berger_gui.py
+61 -9
View File
@@ -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")