commit b390792e34f1ecd3c55ac8e55902191a4e783f48 Author: acidburns Date: Mon Jul 27 01:20:56 2026 +0200 Add Berger LFP BLE command notes and client diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..17cd2fc --- /dev/null +++ b/.gitignore @@ -0,0 +1,4 @@ +.venv/ +__pycache__/ +*.pyc +.pytest_cache/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..99f8e72 --- /dev/null +++ b/README.md @@ -0,0 +1,130 @@ +# Berger LFP BLE Commands + +Reverse-engineering notes for the Berger LFP Android app BLE protocol. + +This repository documents the custom JBD-style BLE GATT protocol used by the +Berger LFP Android app to read battery telemetry such as SOC, current, voltage, +remaining capacity, cell voltages, temperatures, MOS state, and estimated +charge/discharge time. + +## Target App + +- App label: Berger LFP +- Package: `com.jbd.berger` +- Version analyzed: `1.0.3` / `8` +- XAPK source: APKPure direct download for Play package `com.jbd.berger` +- XAPK SHA-256: `962eb7d0c62a34e03bfbe034632fd227e0f352cd66fac9ac3d77166cec0abb5e` +- Base APK SHA-256: `aa2325cfe3777824657f0ade0c381cf4bc296f9630179e0d33ed4eb56d78c1bb` +- Signer CN: `Unknown` +- Signer cert SHA-256: `ec11c6d3260adecb382186a61207e61657834b3b025e3ca31bc6ef4d88f720f9` + +The XAPK was unpacked and the base APK was decompiled with apktool. The app is +a DCloud/uni-app package, so the useful BLE logic is in +`assets/apps/__UNI__F61076D/www/app-service.js`. + +## Main Finding + +Berger LFP does not use the standard BLE Battery Service. It connects as a BLE +GATT client, scans for a custom service `ff00`, subscribes to notifications on +`ff01`, and writes JBD command frames to `ff02`. + +Android pairing/bonding does not appear to be required by the app flow. The +normal path is adapter init, scan, connect, service discovery, notification +enable, and command writes. + +## Quick Start + +Build the main live-data read frame: + +```sh +python3 tools/berger_frame.py read 0x03 +``` + +Expected output: + +```text +DDA50300FFFD77 +``` + +Send that frame to characteristic `0000ff02-0000-1000-8000-00805f9b34fb` +after subscribing to notifications on +`0000ff01-0000-1000-8000-00805f9b34fb`. + +## Local Desktop App + +This repo includes a small Python GUI client that can scan for Berger/JBD-style +BLE devices, connect, poll telemetry, and display decoded values locally. + +### Debian + +```sh +sudo apt update +sudo apt install python3 python3-venv python3-pip python3-tk bluez + +python3 -m venv .venv +. .venv/bin/activate +python3 -m pip install -r requirements.txt +python3 berger_gui.py +``` + +If scanning finds no devices, check that Bluetooth is powered and unblocked: + +```sh +rfkill list bluetooth +bluetoothctl power on +bluetoothctl scan on +``` + +### Windows 11 + +```powershell +py -3 -m venv .venv +.\.venv\Scripts\Activate.ps1 +python -m pip install --upgrade pip +python -m pip install -r requirements.txt +python berger_gui.py +``` + +The GUI uses native WinRT Bluetooth through `bleak` on Windows. + +## App Workflow + +The app's telemetry flow is: + +1. Open the Bluetooth adapter. +2. Start BLE discovery filtered to service `0000ff00-0000-1000-8000-00805f9b34fb`. +3. Connect to the selected peripheral. +4. Discover services and characteristics. +5. Subscribe to notifications on `ff01`. +6. Write JBD read frames to `ff02`. +7. Buffer notify chunks, split complete `DD ... 77` frames, validate checksum, + and parse the response payload into UI state. + +## Minimal BLE Workflow + +1. Connect to the device over BLE GATT. +2. Discover services. +3. Subscribe to notifications on `ff01` under service `ff00`. +4. Write the base-data read frame to `ff02`: + + ```text + DDA50300FFFD77 + ``` + +5. Write the cell-voltage read frame to `ff02`: + + ```text + DDA50400FFFC77 + ``` + +6. Parse notifications from `ff01`. + +## Caveats + +- This catalog is code-derived from the Android app bundle, not yet confirmed + against a live HCI capture. +- The desktop app currently implements read-only telemetry. The Android app has + charge/discharge MOS and factory/name/capacity write paths; those are + documented, but deliberately not exposed as GUI buttons yet. +- Only one phone/client can normally hold the battery GATT connection at a time. + Close the Android app before connecting from a PC. diff --git a/berger_gui.py b/berger_gui.py new file mode 100644 index 0000000..07bb429 --- /dev/null +++ b/berger_gui.py @@ -0,0 +1,365 @@ +#!/usr/bin/env python3 +"""Small desktop BLE client for Berger LFP / JBD-style battery telemetry.""" + +from __future__ import annotations + +import asyncio +import threading +from dataclasses import dataclass, field + +try: + import tkinter as tk + from tkinter import messagebox, ttk +except ModuleNotFoundError: # Allows parser tests on minimal/headless Python. + tk = None + messagebox = None + ttk = None + +try: + from bleak import BleakClient, BleakScanner +except ModuleNotFoundError: # Allows parser tests before GUI dependencies are installed. + BleakClient = None + BleakScanner = None + +SERVICE_UUID = "0000ff00-0000-1000-8000-00805f9b34fb" +NOTIFY_UUID = "0000ff01-0000-1000-8000-00805f9b34fb" +WRITE_UUID = "0000ff02-0000-1000-8000-00805f9b34fb" + + +def checksum(command: int, payload: bytes = b"") -> int: + total = (command + len(payload) + sum(payload)) & 0xFFFF + return ((total ^ 0xFFFF) + 1) & 0xFFFF + + +def build_frame(mode: int, command: int, payload: bytes = b"") -> bytes: + check = checksum(command, payload) + return bytes([0xDD, mode, command, len(payload), *payload, check >> 8, check & 0xFF, 0x77]) + + +def read_frame(command: int) -> bytes: + return build_frame(0xA5, command) + + +def write_frame(command: int, payload: bytes) -> bytes: + return build_frame(0x5A, command, payload) + + +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 + actual = int.from_bytes(frame[-3:-1], "big") + return actual in {expected, response_expected} + + +def split_frames(buffer: bytearray) -> list[bytes]: + frames: list[bytes] = [] + while True: + try: + start = buffer.index(0xDD) + except ValueError: + buffer.clear() + return frames + if start: + del buffer[:start] + if len(buffer) < 7: + return frames + frame_len = buffer[3] + 7 + if len(buffer) < frame_len: + return frames + frame = bytes(buffer[:frame_len]) + del buffer[:frame_len] + if frame[-1] == 0x77: + frames.append(frame) + + +@dataclass +class Telemetry: + voltage: float = 0.0 + current: float = 0.0 + power: float = 0.0 + soc: int = 0 + remaining_capacity: float = 0.0 + nominal_capacity: float = 0.0 + cycle_count: int = 0 + production_date: str = "" + software_version: str = "" + charge_mos: bool = False + discharge_mos: bool = False + balancing: bool = False + protection: bool = False + protection_bit: int | None = None + temperatures: list[float] = field(default_factory=list) + cells: list[float] = field(default_factory=list) + time_label: str = "Est. time" + time_value: str = "--" + + +@dataclass +class DeviceState: + telemetry: Telemetry = field(default_factory=Telemetry) + rx_log: list[str] = field(default_factory=list) + tx_log: list[str] = field(default_factory=list) + + +def word(data: bytes, offset: int) -> int: + return (data[offset] << 8) | data[offset + 1] + + +def signed_current(raw: int) -> float: + if raw > 32768: + raw -= 65536 + return round(raw / 100, 2) + + +def transform_time(hours: float) -> str: + whole = int(hours) + minutes = int((hours - whole) * 60) + return f"{whole} h {minutes:02d} m" + + +def decode_response(frame: bytes, state: DeviceState) -> None: + if not validate_frame(frame): + return + command = frame[1] + status = frame[2] + length = frame[3] + payload = frame[4 : 4 + length] + t = state.telemetry + + if command == 0x03 and status == 0 and length >= 23: + t.voltage = round(word(payload, 0) / 100, 2) + t.current = signed_current(word(payload, 2)) + t.power = round(t.voltage * t.current, 2) + t.remaining_capacity = round(word(payload, 4) / 100, 2) + t.nominal_capacity = round(word(payload, 6) / 100, 2) + t.cycle_count = word(payload, 8) + date_raw = word(payload, 10) + day = date_raw & 0x1F + month = (date_raw >> 5) & 0x0F + year = 2000 + (date_raw >> 9) + t.production_date = f"{year}-{month}-{day}" + t.balancing = payload[12] != 0 or payload[13] != 0 + t.protection = payload[16] != 0 or payload[17] != 0 + if t.protection: + bits = f"{payload[16]:08b}{payload[17]:08b}"[::-1] + t.protection_bit = bits.find("1") + version = str(payload[18]) + t.software_version = f"{version[0]}.{version[1]}" if len(version) > 1 else version + t.soc = payload[19] + mos = payload[20] + t.charge_mos = mos in {1, 3} + t.discharge_mos = mos in {2, 3} + temp_count = payload[22] if length > 22 else 0 + temps = [] + for idx in range(temp_count): + off = 23 + idx * 2 + if off + 1 < len(payload): + temps.append(round((word(payload, off) - 2731) / 10, 1)) + t.temperatures = temps + if t.current == 0: + t.time_label = "Est. time" + t.time_value = "--" + elif t.current > 0 and t.nominal_capacity: + t.time_label = "Time till full" + t.time_value = transform_time((t.nominal_capacity - t.remaining_capacity) / t.current) + elif t.current < 0: + t.time_label = "Time till empty" + t.time_value = transform_time(t.remaining_capacity / -t.current) + + elif command == 0x04 and status == 0: + cells = [] + for off in range(0, len(payload) - 1, 2): + cells.append(round(word(payload, off) / 1000, 3)) + t.cells = cells + + +def berger_match_reason(name: str | None, service_uuids: list[str]) -> str | None: + normalized = {s.lower() for s in service_uuids} + if SERVICE_UUID in normalized: + return "service:ff00" + if name and any(token in name.upper() for token in ("JBD", "BMS", "BERGER")): + return "name" + return None + + +class BergerApp: + def __init__(self, root: tk.Tk) -> None: + self.root = root + self.root.title("Berger LFP BLE") + self.state = DeviceState() + self.devices = [] + self.client: BleakClient | None = None + self.buffer = bytearray() + 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.rows: dict[str, tk.StringVar] = {} + self._build_ui() + + def _build_ui(self) -> None: + root = ttk.Frame(self.root, padding=10) + root.grid(sticky="nsew") + self.root.columnconfigure(0, weight=1) + self.root.rowconfigure(0, weight=1) + + top = ttk.Frame(root) + top.grid(row=0, column=0, sticky="ew") + ttk.Button(top, text="Scan", command=self.scan).grid(row=0, column=0) + ttk.Spinbox(top, from_=3, to=20, textvariable=self.scan_seconds, width=4).grid(row=0, column=1) + ttk.Entry(top, textvariable=self.direct, width=32).grid(row=0, column=2, padx=6) + ttk.Button(top, text="Connect", command=self.connect_selected).grid(row=0, column=3) + ttk.Label(top, textvariable=self.status).grid(row=0, column=4, padx=8, sticky="w") + + self.listbox = tk.Listbox(root, height=8) + self.listbox.grid(row=1, column=0, sticky="nsew", pady=8) + + data = ttk.Frame(root) + data.grid(row=2, column=0, sticky="nsew") + labels = [ + ("soc", "SOC"), + ("voltage", "Voltage"), + ("current", "Current"), + ("power", "Power"), + ("remaining_capacity", "Remaining Ah"), + ("nominal_capacity", "Nominal Ah"), + ("cycle_count", "Cycles"), + ("mos", "MOS"), + ("temperatures", "Temps"), + ("cells", "Cells"), + ("time", "Time"), + ("production_date", "Production"), + ("software_version", "Software"), + ("status", "Status"), + ] + for idx, (key, label) in enumerate(labels): + self.rows[key] = tk.StringVar(value="--") + 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") + + controls = ttk.Frame(root) + controls.grid(row=3, 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) + + def run(self, coro): + return asyncio.run_coroutine_threadsafe(coro, self.loop) + + def scan(self) -> None: + self.status.set("Scanning") + self.run(self._scan()) + + async def _scan(self) -> None: + if BleakScanner is None: + self.root.after(0, lambda: self.status.set("Install bleak first")) + return + devices = await BleakScanner.discover(timeout=self.scan_seconds.get(), return_adv=True) + self.devices = [] + for device, adv in devices.values(): + services = list(adv.service_uuids or []) + reason = berger_match_reason(device.name, services) + self.devices.append((device, adv, reason)) + self.devices.sort(key=lambda item: (item[2] is None, -(item[1].rssi or -999))) + self.root.after(0, self._render_devices) + + def _render_devices(self) -> None: + self.listbox.delete(0, tk.END) + for device, adv, reason in self.devices: + mark = "*" if reason else " " + self.listbox.insert(tk.END, f"{mark} {device.address} {adv.rssi:>4} {device.name or ''} {reason or ''}") + self.status.set(f"{len(self.devices)} devices") + + def connect_selected(self) -> None: + address = self.direct.get().strip() + if not address: + selected = self.listbox.curselection() + if not selected: + messagebox.showinfo("Berger LFP BLE", "Select a device or enter an address.") + return + address = self.devices[selected[0]][0].address + self.status.set("Connecting") + self.run(self._connect(address)) + + async def _connect(self, address: str) -> None: + if BleakClient is None: + self.root.after(0, lambda: self.status.set("Install bleak first")) + return + self.client = BleakClient(address) + await self.client.connect() + 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")) + + 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) + + def _on_notify(self, _sender, data: bytearray) -> None: + self.buffer.extend(data) + 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) + + def _render_state(self) -> None: + t = self.state.telemetry + self.rows["soc"].set(f"{t.soc} %") + self.rows["voltage"].set(f"{t.voltage:.2f} V") + self.rows["current"].set(f"{t.current:.2f} A") + self.rows["power"].set(f"{t.power:.2f} W") + self.rows["remaining_capacity"].set(f"{t.remaining_capacity:.2f} Ah") + 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.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["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 "--") + status = [] + if t.balancing: + status.append("balancing") + if t.protection: + status.append(f"protection bit {t.protection_bit}") + self.rows["status"].set(", ".join(status) or "normal") + + def poll_once(self) -> None: + self.run(self._poll_once()) + + async def _poll_once(self) -> None: + await self._send(read_frame(0x03)) + await self._send(read_frame(0x04)) + + def disconnect(self) -> None: + self.run(self._disconnect()) + + async def _disconnect(self) -> None: + if self.client: + await self.client.disconnect() + self.client = None + self.root.after(0, lambda: self.status.set("Disconnected")) + + +def main() -> None: + if tk is None: + raise SystemExit("tkinter is not installed. Install python3-tk to run the GUI.") + root = tk.Tk() + BergerApp(root) + root.mainloop() + + +if __name__ == "__main__": + main() diff --git a/docs/ble-flow.md b/docs/ble-flow.md new file mode 100644 index 0000000..9e68f94 --- /dev/null +++ b/docs/ble-flow.md @@ -0,0 +1,105 @@ +# Berger LFP BLE Flow + +This is the app flow used to obtain battery information such as discharge +current, SOC, voltage, temperatures, cell voltages, MOS state, and estimated +charge/discharge time. + +## Evidence + +- App package: `com.jbd.berger` +- Version: `1.0.3` +- Main bundle: `assets/apps/__UNI__F61076D/www/app-service.js` +- Source-map style log labels embedded in bundle: + - `utils/BLE.ts` + - `App.vue` + - dashboard/control/parameter pages + +## Connection Sequence + +1. `uni.openBluetoothAdapter()` +2. `uni.startBluetoothDevicesDiscovery()` with service filter `ff00` +3. `uni.onBluetoothDeviceFound()` records devices and RSSI +4. `uni.createBLEConnection()` with 10 second timeout +5. `uni.getBLEDeviceServices()` +6. `uni.getBLEDeviceCharacteristics()` +7. `uni.notifyBLECharacteristicValueChange()` on `ff01` +8. `uni.onBLECharacteristicValueChange()` buffers incoming bytes +9. `uni.writeBLECharacteristicValue()` sends command chunks to `ff02` with + `writeType: "writeNoResponse"` + +The app chunks outgoing buffers into 20-byte pieces before writing. + +## Services And Characteristics + +- Service: `0000ff00-0000-1000-8000-00805f9b34fb` +- Notify/read: `0000ff01-0000-1000-8000-00805f9b34fb` +- Write: `0000ff02-0000-1000-8000-00805f9b34fb` + +## Pairing / Bonding + +- Needs Android bond before normal telemetry: likely no. +- Evidence: the app uses uni-app BLE APIs for connection, service discovery, + notifications, and writes. I did not find a normal-path bond request in the + app bundle. +- Caveat: a peripheral firmware could still enforce encryption independently, + but the app flow itself does not show pairing as a telemetry prerequisite. + +## Getting Live Values + +The app sends: + +```text +DD A5 03 00 FF FD 77 +``` + +The response uses: + +```text +DD 03 77 +``` + +Important payload fields for command `0x03`: + +| Payload bytes | Formula | Unit | Field | +| ---: | --- | --- | --- | +| `0..1` | big-endian / 100 | V | pack voltage | +| `2..3` | signed big-endian / 100 | A | current | +| `4..5` | big-endian / 100 | Ah | remaining capacity | +| `6..7` | big-endian / 100 | Ah | nominal capacity | +| `8..9` | raw | cycles | cycle count | +| `10..11` | JBD bitfield | date | production date | +| `12..13` | nonzero bitmap | boolean | balancing | +| `16..17` | nonzero bitmap | boolean | protection | +| `18` | decimal digits | version | software version | +| `19` | raw | % | SOC | +| `20` | `1 = charge`, `2 = discharge`, `3 = both` | boolean | MOS state | +| `22` | raw count | count | temperature sensor count | +| `23..` | `(raw - 2731) / 10` | deg C | temperatures | + +Production date bitfield: + +```text +day = value & 0x1F +month = (value >> 5) & 0x0F +year = 2000 + (value >> 9) +``` + +Estimated time is computed locally: + +```text +if current_A > 0: + time_to_full = (nominal_Ah - remaining_Ah) / current_A +if current_A < 0: + time_to_empty = remaining_Ah / abs(current_A) +``` + +## Cell Voltages + +The app sends: + +```text +DD A5 04 00 FF FC 77 +``` + +The response command `0x04` payload is a sequence of big-endian 16-bit cell +voltages scaled by `0.001 V`. diff --git a/docs/command-catalog.md b/docs/command-catalog.md new file mode 100644 index 0000000..2858863 --- /dev/null +++ b/docs/command-catalog.md @@ -0,0 +1,80 @@ +# Berger LFP BLE Command Catalog + +Status: decompiled from Berger LFP Android app `1.0.3`. Entries are based on +the bundled uni-app JavaScript code, not live BLE captures. + +## GATT Surface + +| UUID | Direction | Purpose | +| --- | --- | --- | +| `0000ff00-0000-1000-8000-00805f9b34fb` | service | Main JBD/BMS service | +| `0000ff01-0000-1000-8000-00805f9b34fb` | notify/read | BMS response frames | +| `0000ff02-0000-1000-8000-00805f9b34fb` | write no response | BMS command frames | + +## Frame Format + +Read command: + +```text +DD A5 77 +``` + +Write command: + +```text +DD 5A 77 +``` + +Checksum: + +```text +sum = command + length + sum(payload) +checksum = ((sum ^ 0xFFFF) + 1) & 0xFFFF +``` + +Responses observed in the parser: + +```text +DD 77 +``` + +## Commands In App + +| Name | Frame / payload | Meaning | +| --- | --- | --- | +| Read base data | `DD A5 03 00 FF FD 77` | Pack voltage/current/SOC/capacity/MOS/temps | +| Read cell voltages | `DD A5 04 00 FF FC 77` | Per-cell voltage list | +| Read hardware version | `DD A5 05 00 FF FB 77` | ASCII hardware version | +| Enter factory mode | `DD 5A 00 02 56 78 FF 30 77` | Factory/settings mode | +| Exit factory mode | `DD 5D 01 02 28 28 FF AD 77` | Exit factory/settings mode | +| Read BMS model | `DD A5 FA 03 00 B0 04 FE 4F 77` | New-style FA parameter `0xB0`, length `4` | +| Read battery model | `DD A5 FA 03 00 9E 0C FE 59 77` | New-style FA parameter `0x9E`, length `12` | +| Read full-charge capacity | `DD A5 FA 03 00 70 01 FE 92 77` | New-style FA parameter `0x70`, length `1` | +| Read barcode, new version | `DD A5 FA 03 00 58 10 FE 95 77` | New-style FA parameter `0x58`, length `16` | +| Read manufacturer, new version | `DD A5 FA 03 00 38 10 FE B5 77` | New-style FA parameter `0x38`, length `16` | +| Read barcode, old version | `DD A5 A2 00 FF 5E 77` | Old-style barcode | +| Read manufacturer, old version | `DD A5 A0 00 FF 60 77` | Old-style manufacturer | +| Set charge/discharge MOS | `DD 5A E1 02 00 77` | Mode `0..3`, see below | +| Set BLE name | `FF AA 07 ` | Alternate name command frame | + +## MOS Control Mode + +The Android app writes command `0xE1` with payload `00 `. + +| Mode | Meaning inferred from app logic | +| ---: | --- | +| `0` | charge on, discharge on | +| `1` | charge off, discharge on | +| `2` | charge on, discharge off | +| `3` | charge off, discharge off | + +The app response handler treats command `0xE1` status `0` as success. + +## Scan Hints + +The app scans using only the service filter `ff00`. It also derives a MAC-like +address from advertising data on iOS and reverses byte order when the derived +address ends in `A5`, `A4`, `12`, `52`, or `00`. + +The desktop utility treats service `ff00` as the strong match and device names +containing `JBD`, `BMS`, or `BERGER` as soft hints. diff --git a/evidence/apk-info.md b/evidence/apk-info.md new file mode 100644 index 0000000..69c909b --- /dev/null +++ b/evidence/apk-info.md @@ -0,0 +1,43 @@ +# APK Evidence + +- File analyzed locally: `Berger_LFP_1.0.3_APKPure.xapk` +- Source: APKPure direct XAPK download +- Play URL: `https://play.google.com/store/apps/details?id=com.jbd.berger&hl=de` +- App package: `com.jbd.berger` +- App label: `Berger LFP` +- Publisher shown by APKPure/Play metadata: `Fritz Berger GmbH` +- Version: `1.0.3` / `8` +- XAPK SHA-256: `962eb7d0c62a34e03bfbe034632fd227e0f352cd66fac9ac3d77166cec0abb5e` +- Base APK SHA-256: `aa2325cfe3777824657f0ade0c381cf4bc296f9630179e0d33ed4eb56d78c1bb` +- Signer DN: `CN=Unknown, OU=Unknown, O=Unknown, L=Unknown, ST=Unknown, C=Unknown` +- Signer cert SHA-256: `ec11c6d3260adecb382186a61207e61657834b3b025e3ca31bc6ef4d88f720f9` +- Source stamp signer DN: `CN=Android, OU=Android, O=Google Inc., L=Mountain View, ST=California, C=US` +- Source stamp signer SHA-256: `3257d599a49d2c961a471ca9843f59d341a405884583fc087df4237b733bbd6d` + +The XAPK contains: + +- `com.jbd.berger.apk` +- `config.armeabi_v7a.apk` +- language splits: `en`, `zh` +- density split: `config.mdpi.apk` +- `manifest.json` +- `icon.png` + +Split/package SHA-256: + +| File | SHA-256 | +| --- | --- | +| `com.jbd.berger.apk` | `aa2325cfe3777824657f0ade0c381cf4bc296f9630179e0d33ed4eb56d78c1bb` | +| `config.armeabi_v7a.apk` | `fc3006094ca26f1b3e51b8848f7c2e2c9225c0fb10aafd08214845bd1e262d1b` | +| `config.en.apk` | `bf9472ba9cb732827ec1483287cca9598614f1f4daa20b905e494cd49eaba3da` | +| `config.mdpi.apk` | `45546c07728a3ffb05c9c7d42418160adeda29466000f813d8b17747073da43d` | +| `config.zh.apk` | `f1c18a5f380a45f7175baa7d646f15b774ede0d582ee436d7fafc86f456cab67` | +| `manifest.json` | `9a35b17653ed11342cc93e2abc81f3a7698632def42ec9a0644f5eb62848255d` | + +Local analysis outputs: + +- XAPK unpack: `berger-ble-work/xapk` +- apktool output: `berger-ble-work/apktool-out` +- Main app bundle: `berger-ble-work/apktool-out/assets/apps/__UNI__F61076D/www/app-service.js` + +The APK/XAPK files are intentionally not committed into this repository. diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..af8eee2 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +bleak>=0.22 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/test_berger_gui.py b/tests/test_berger_gui.py new file mode 100644 index 0000000..a0a8da2 --- /dev/null +++ b/tests/test_berger_gui.py @@ -0,0 +1,70 @@ +import unittest + +from berger_gui import DeviceState, berger_match_reason, decode_response, read_frame, split_frames, write_frame + + +class BergerParserTest(unittest.TestCase): + def test_read_frame_checksum(self): + self.assertEqual(read_frame(0x03).hex().upper(), "DDA50300FFFD77") + self.assertEqual(read_frame(0x04).hex().upper(), "DDA50400FFFC77") + + def test_write_switch_checksum(self): + self.assertEqual(write_frame(0xE1, b"\x00\x00").hex().upper(), "DD5AE1020000FF1D77") + + def test_decode_base_data(self): + payload = bytearray(27 + 4) + payload[0:2] = (1328).to_bytes(2, "big") + payload[2:4] = (65536 - 752).to_bytes(2, "big") + payload[4:6] = (10750).to_bytes(2, "big") + payload[6:8] = (16000).to_bytes(2, "big") + payload[8:10] = (3).to_bytes(2, "big") + payload[10:12] = ((25 << 9) | (7 << 5) | 27).to_bytes(2, "big") + payload[16:18] = (0x0010).to_bytes(2, "big") + payload[18] = 21 + payload[19] = 67 + payload[20] = 3 + 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" + + buffer = bytearray(b"\x00" + frame + frame) + frames = split_frames(buffer) + self.assertEqual(len(frames), 2) + + state = DeviceState() + decode_response(frames[0], state) + self.assertAlmostEqual(state.telemetry.voltage, 13.28) + self.assertAlmostEqual(state.telemetry.current, -7.52) + self.assertEqual(state.telemetry.soc, 67) + self.assertAlmostEqual(state.telemetry.remaining_capacity, 107.5) + self.assertAlmostEqual(state.telemetry.nominal_capacity, 160.0) + self.assertEqual(state.telemetry.cycle_count, 3) + self.assertEqual(state.telemetry.production_date, "2025-7-27") + self.assertTrue(state.telemetry.charge_mos) + self.assertTrue(state.telemetry.discharge_mos) + self.assertTrue(state.telemetry.protection) + self.assertEqual(state.telemetry.temperatures, [25.0, 26.0]) + self.assertEqual(state.telemetry.time_label, "Time till empty") + + 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" + state = DeviceState() + decode_response(frame, state) + self.assertEqual(state.telemetry.cells, [3.306, 3.312, 3.308, 3.311]) + + 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") + self.assertIsNone(berger_match_reason("Keyboard", [])) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/berger_frame.py b/tools/berger_frame.py new file mode 100644 index 0000000..b0df0d5 --- /dev/null +++ b/tools/berger_frame.py @@ -0,0 +1,62 @@ +#!/usr/bin/env python3 +"""Build Berger/JBD BLE frames observed in the Android app.""" + +from __future__ import annotations + +import argparse + + +def checksum(command: int, payload: bytes = b"") -> int: + """JBD two's-complement checksum over command, length, and payload.""" + total = (command + len(payload) + sum(payload)) & 0xFFFF + return ((total ^ 0xFFFF) + 1) & 0xFFFF + + +def frame(mode: int, command: int, payload: bytes = b"") -> bytes: + check = checksum(command, payload) + return bytes([0xDD, mode, command, len(payload), *payload, check >> 8, check & 0xFF, 0x77]) + + +def read_frame(command: int) -> bytes: + return frame(0xA5, command) + + +def write_frame(command: int, payload: bytes) -> bytes: + return frame(0x5A, command, payload) + + +def name_frame(command: int, payload: bytes = b"") -> bytes: + check = (command + len(payload) + sum(payload)) & 0xFF + return bytes([0xFF, 0xAA, command, len(payload), *payload, check]) + + +def parse_hex_bytes(value: str) -> bytes: + clean = value.replace(" ", "").replace(":", "") + return bytes.fromhex(clean) + + +def main() -> None: + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="cmd", required=True) + + p_read = sub.add_parser("read") + p_read.add_argument("command", type=lambda x: int(x, 0)) + + p_write = sub.add_parser("write") + p_write.add_argument("command", type=lambda x: int(x, 0)) + p_write.add_argument("payload_hex", nargs="?", default="") + + p_name = sub.add_parser("name") + p_name.add_argument("value") + + args = parser.parse_args() + if args.cmd == "read": + print(read_frame(args.command).hex().upper()) + elif args.cmd == "write": + print(write_frame(args.command, parse_hex_bytes(args.payload_hex)).hex().upper()) + elif args.cmd == "name": + print(name_frame(0x07, args.value.encode("ascii")).hex().upper()) + + +if __name__ == "__main__": + main()