Guten Tag
Habe in meinem Van ein Accubox von Ective verbaut.
Kann diese auch mit der zugehörigen App über Bluetooth auslesen, würde es aber gerne in den VanPI Integrieren.
Wer kann helfen?
Danke
Hallo Klaus, ich bin mal von dem anderen in deinen Topic rübergekommen da er spezifischer auf die Accubox von Ective ausgelegt ist. Du hattest in dem anderen Topic gefragt wie man das Skript zum Laufen bekommt. Erstmal kurz zum Stand des Skripts, es ist derzeit hier am werkeln und ich bekomme tatsächlich gute Daten aus der Accubox, es gibt gelegentliche einzelne Ausreisser in der Ausgabe was sich aber bis jetzt immer wieder eingependelt hat. Ich dachte mir ich stelle es mal zu Verfügung da es hier im Forum den ein oder anderen gab der Interesse daran hatte und da ich natürlich sehr an Verbesserungen interessiert bin dachte ich mir, vielleicht kann man es hier gemeinsam mal testen und schauen ob und wie gut es funktioniert.
Kurz zu mir, ich habe einige Produkte von Peakaway (Fussy und die Hochstrombüchse 
) aber kein Core, Weihnachten dauert noch ein bisschen 
. Ich habe aber ebenfalls einen Raspberry Pi 4B mit einem Raspbian Lite Version “Debian GNU/Linux 12 (bookworm)” im Camper am laufen der ein Dashboard und ein Cockpit über das lokale WLAN (Pi mittels LAN an einen GL.Inet angebunden) zu Verfügung stellt.
Das Skript: Zum testen und zum entwickeln habe ich erstmal einfach den Laptop genommen und die wirklich langsam werdende Entwicklung irgendwann auf den Pi verschoben, wichtig ist das du einen Bluetooth Adapter an Board hast der BLE unterstützt sollte eigentlich ab in und um 2015 Standart sein. Ich würde an deiner Stelle auch erstmal über den Laptop schauen ob das Skript auch für dein Baujahr der Accubox funktioniert.
MAC Adresse ausfindig machen: Die MAC sollte eigentlich auf der Accubox zufinden sein, die brauchst du später für das Skript da meine derzeit dort hardgecodet drinne steht
# Default MAC address (can be overridden with --mac)
DEFAULT_MAC = "58:b6:4f:49:9c:f1"
musst du entweder das --mac [deine MAC Adresse} flag im Kommando nehmen, das überschreibt die hardgecodete MAC oder du trägst im Skript direkt deine an obiger Stelle im Code ein.
Du brauchst bleak – das ist eine kleine Python-Bibliothek, die sich um die Bluetooth Low Energy Verbindung zu deiner AccuBox kümmert (scannen, verbinden, Daten auslesen), installieren kannst du sie über pip z.b. so:
bash
pip install --user bleak
was die einzige dependency ist. Danach einmal aus- und wieder einloggen (damit die Bluetooth-Berechtigung greift).
Nebenbei: Wie die Daten dann im VanPi weiterverarbeitet werden, weiß ich selbst nicht genau (wie schon geschreiben, habe leider kein VanPi System hier) aber nach meiner Recherche läuft das wahrscheinlich über MQTT ? Ein bisschen vorbereitend (sofern es überhaupt gebrauch finden kann), habe ich das Skript schon mal um eine MQTT-Publish-Funktion erweitert. Für den ersten Test reicht es aber völlig, wenn wir erstmal nur sehen, ob überhaupt Daten von deiner Batterie kommen.
Du kannst dieses Skript ja mal testen
#!/usr/bin/env python3
"""
AccuBox BLE Reader for Ective AccuBox (S200)
This script connects to an Ective AccuBox battery via Bluetooth Low Energy,
sends a status request, decodes the 21‑byte notification and outputs the
battery parameters as JSON. It supports both one‑time and cyclic polling,
as well as optional MQTT publishing using `mosquitto_pub` (no Python MQTT
library required).
Features:
- One‑time or cyclic query mode
- Automatic charge/discharge mode detection
- MQTT publishing with configurable broker, topic, QoS, retain, fanout
- Raw hex output for debugging
- Graceful shutdown on SIGINT/SIGTERM
Requirements:
- Python 3.10+
- bleak library (install with `pip install --user bleak`)
- mosquitto-clients (only for MQTT, `sudo apt install mosquitto-clients`)
Usage examples:
# One‑time read (using default MAC or custom)
python3 script.py
python3 script.py --mac 58:b6:4f:49:9c:f1
# Cyclic every 10 seconds
python3 script.py --mode cyclic --interval 10
# Cyclic + MQTT (custom broker and topic)
python3 script.py --mode cyclic --interval 10 --publish mosquitto --topic my/battery --mqtt-host 192.168.1.100
# With fanout (individual subtopics for each field)
python3 script.py --mode cyclic --publish mosquitto --fanout
Author: ummeegge / Erik
License: MIT
"""
from __future__ import annotations
import argparse
import asyncio
import json
import logging
import subprocess
import sys
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from shutil import which
from typing import Optional
from bleak import BleakClient, BleakScanner
# ---------- BLE protocol constants ----------
# Default MAC address (can be overridden with --mac)
DEFAULT_MAC = "58:b6:4f:49:9c:f1"
CHAR_UUID = "0000ffe1-0000-1000-8000-00805f9b34fb"
CMD_REQUEST_STATUS = bytes.fromhex("1c030400000846b1")
# Empirical scaling factors (from reverse engineering)
VOLT_SCALING_CHARGE = 47.3 # voltage raw > 200 → charging
VOLT_SCALING_DISCHARGE = 10.0 # voltage raw < 200 → discharging
CURRENT_SCALING_CHARGE = 94.0 # current from bytes 12‑13 (charging)
CURRENT_SCALING_DISCHARGE = 7965.0 # current from bytes 18‑19 (discharging)
LOGGER = logging.getLogger("accubox")
# ---------- Data model ----------
@dataclass
class Status:
"""Decoded battery status data."""
status: str # "ok" or "error"
timestamp: str # ISO format with timezone
capacity_ah: Optional[int] = None
soc_percent: Optional[int] = None
rest_time: Optional[str] = None # "hh:mm"
voltage_v: Optional[float] = None
current_a: Optional[float] = None
power_w: Optional[float] = None
mode: Optional[str] = None # "charging" or "discharging"
raw_hex: Optional[str] = None # full 21‑byte hex dump (if --raw)
message: Optional[str] = None # error message
# ---------- Helper functions ----------
def utc_now_iso() -> str:
"""Return current local time in ISO format with timezone."""
return datetime.now(timezone.utc).astimezone().isoformat()
def status_to_payload(status: Status) -> str:
"""Serialize a Status object to a JSON string."""
return json.dumps(asdict(status), ensure_ascii=False)
def decode_status(data: bytes, include_raw: bool = True) -> Status:
"""
Decode a 21‑byte BLE notification into a Status object.
Args:
data: Raw bytes received from the characteristic.
include_raw: If True, add the hexadecimal representation to the status.
Returns:
Status object with decoded fields.
"""
if len(data) != 21:
return Status(
status="error",
timestamp=utc_now_iso(),
message=f"Unexpected length: {len(data)}",
raw_hex=data.hex() if include_raw else None,
)
# Extract little‑endian fields
capacity_ah = int.from_bytes(data[4:6], "little")
soc_percent = int.from_bytes(data[6:8], "little")
rest_hours = int.from_bytes(data[8:10], "little")
rest_minutes = int.from_bytes(data[10:12], "little")
current_charge_raw = int.from_bytes(data[12:14], "little", signed=True)
# bytes 14-16 are temperature (unused, not displayed in app)
voltage_raw = int.from_bytes(data[16:18], "little")
current_discharge_raw = int.from_bytes(data[18:20], "little", signed=True)
# Mode detection based on voltage raw value
if voltage_raw > 200:
mode = "charging"
voltage_v = round(voltage_raw / VOLT_SCALING_CHARGE, 2)
current_a = round(abs(current_charge_raw) / CURRENT_SCALING_CHARGE, 2)
else:
mode = "discharging"
voltage_v = round(voltage_raw / VOLT_SCALING_DISCHARGE, 2)
current_a = -round(abs(current_discharge_raw) / CURRENT_SCALING_DISCHARGE, 2)
power_w = round(abs(current_a) * voltage_v, 1)
if current_a < 0:
power_w = -power_w
return Status(
status="ok",
timestamp=utc_now_iso(),
capacity_ah=capacity_ah,
soc_percent=soc_percent,
rest_time=f"{rest_hours:02d}:{rest_minutes:02d}",
voltage_v=voltage_v,
current_a=current_a,
power_w=power_w,
mode=mode,
raw_hex=data.hex() if include_raw else None,
)
# ---------- Command line argument parser ----------
def build_parser() -> argparse.ArgumentParser:
"""Build and return the argument parser with all options."""
parser = argparse.ArgumentParser(
description="Ective AccuBox BLE to JSON/MQTT reader",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument(
"--mac",
default=DEFAULT_MAC,
help="BLE MAC address (default: %(default)s)",
)
parser.add_argument(
"--mode",
choices=["once", "cyclic"],
default="once",
help="Query mode: once or cyclic",
)
parser.add_argument(
"--interval",
type=float,
default=5.0,
help="Interval in seconds for cyclic mode",
)
parser.add_argument(
"--timeout",
type=float,
default=3.0,
help="Seconds to wait for a BLE notification",
)
parser.add_argument(
"--verbose", action="store_true", help="Enable debug logging"
)
parser.add_argument(
"--raw", action="store_true", help="Include raw_hex field in output"
)
parser.add_argument(
"--publish",
choices=["none", "stdout", "mosquitto"],
default="none",
help="Where to publish the status",
)
parser.add_argument(
"--topic",
default="battery/accubox",
help="MQTT topic (default: %(default)s)",
)
parser.add_argument(
"--state-topic",
default="battery/accubox/state",
help="MQTT online/offline topic (default: %(default)s)",
)
parser.add_argument(
"--qos", type=int, choices=[0, 1, 2], default=1, help="MQTT QoS level"
)
parser.add_argument(
"--retain", action="store_true", help="Retain MQTT messages"
)
parser.add_argument(
"--no-retain", action="store_true", help="Disable retain"
)
parser.add_argument(
"--fanout",
action="store_true",
help="Publish individual fields as subtopics",
)
parser.add_argument("--mqtt-host", default="localhost", help="MQTT broker host")
parser.add_argument("--mqtt-port", type=int, default=1883, help="MQTT broker port")
parser.add_argument("--mqtt-username", default=None, help="MQTT username (if needed)")
parser.add_argument("--mqtt-password", default=None, help="MQTT password (if needed)")
parser.add_argument(
"--once",
action="store_true",
help="Run exactly one cycle even in cyclic mode",
)
return parser
# ---------- Logging ----------
def setup_logging(verbose: bool) -> None:
"""Configure logging level and format."""
level = logging.DEBUG if verbose else logging.INFO
logging.basicConfig(
level=level, format="%(asctime)s %(levelname)s %(message)s"
)
# ---------- MQTT helpers (using mosquitto_pub) ----------
def effective_retain(args) -> bool:
"""Determine whether to set the MQTT retain flag."""
if args.no_retain:
return False
if args.retain:
return True
return True if args.publish == "mosquitto" else False
def publish_stdout(status: Status) -> None:
"""Print the status as JSON to standard output."""
print(status_to_payload(status), flush=True)
def publish_mosquitto(status: Status, args) -> None:
"""
Publish the status to an MQTT broker using mosquitto_pub.
Args:
status: Status object to publish.
args: Parsed command line arguments with MQTT settings.
Raises:
RuntimeError: If mosquitto_pub is not found in PATH.
subprocess.CalledProcessError: If mosquitto_pub fails.
"""
if which("mosquitto_pub") is None:
raise RuntimeError("mosquitto_pub not found – install mosquitto-clients")
payload = status_to_payload(status)
base = [
"mosquitto_pub",
"-h", args.mqtt_host,
"-p", str(args.mqtt_port),
"-t", args.topic,
"-m", payload,
"-q", str(args.qos),
]
if effective_retain(args):
base.append("-r")
if args.mqtt_username:
base.extend(["-u", args.mqtt_username])
if args.mqtt_password:
base.extend(["-P", args.mqtt_password])
subprocess.run(base, check=True)
# Fanout: one subtopic per field
if args.fanout and status.status == "ok":
fanout_fields = {
"voltage_v": status.voltage_v,
"current_a": status.current_a,
"power_w": status.power_w,
"soc_percent": status.soc_percent,
"capacity_ah": status.capacity_ah,
"mode": status.mode,
"rest_time": status.rest_time,
}
for name, value in fanout_fields.items():
if value is None:
continue
cmd = base.copy()
# replace topic and message
topic_idx = cmd.index("-t") + 1
cmd[topic_idx] = f"{args.topic.rstrip('/')}/{name}"
msg_idx = cmd.index("-m") + 1
cmd[msg_idx] = str(value)
subprocess.run(cmd, check=True)
# State online (always retained)
state_cmd = base.copy()
state_topic_idx = state_cmd.index("-t") + 1
state_cmd[state_topic_idx] = args.state_topic
state_msg_idx = state_cmd.index("-m") + 1
state_cmd[state_msg_idx] = "online"
if "-r" not in state_cmd:
state_cmd.append("-r")
subprocess.run(state_cmd, check=True)
def publish_offline(args) -> None:
"""Publish offline state to the MQTT state topic."""
if which("mosquitto_pub") is None:
return
cmd = [
"mosquitto_pub",
"-h", args.mqtt_host,
"-p", str(args.mqtt_port),
"-t", args.state_topic,
"-m", "offline",
"-q", str(args.qos),
"-r",
]
if args.mqtt_username:
cmd.extend(["-u", args.mqtt_username])
if args.mqtt_password:
cmd.extend(["-P", args.mqtt_password])
try:
subprocess.run(cmd, check=False)
except Exception:
pass
# ---------- BLE communication ----------
async def find_characteristic(client: BleakClient, uuid: str):
"""Return the first GATT characteristic matching the UUID."""
services = client.services if getattr(client, "services", None) else await client.get_services()
for service in services:
for char in service.characteristics:
if char.uuid.lower() == uuid.lower():
return char
return None
async def read_once(mac: str, timeout: float, include_raw: bool) -> Status:
"""
Connect to the battery, request one status update, and return it.
Args:
mac: BLE MAC address.
timeout: Seconds to wait for the notification.
include_raw: Whether to include raw_hex in the output.
Returns:
Status object (may be error status).
"""
device = await BleakScanner.find_device_by_address(mac, timeout=8.0)
if not device:
return Status(status="error", timestamp=utc_now_iso(), message=f"Device {mac} not found")
async with BleakClient(device) as client:
char = await find_characteristic(client, CHAR_UUID)
if not char:
return Status(status="error", timestamp=utc_now_iso(), message="Characteristic not found")
result: Optional[Status] = None
def handler(sender, data):
nonlocal result
result = decode_status(data, include_raw=include_raw)
await client.start_notify(char, handler)
await asyncio.sleep(0.2)
await client.write_gatt_char(char, CMD_REQUEST_STATUS, response=False)
# Simple polling loop (works reliably)
for _ in range(int(timeout * 10)):
if result is not None:
break
await asyncio.sleep(0.1)
await client.stop_notify(char)
return result or Status(status="error", timestamp=utc_now_iso(), message="Timeout")
async def cyclic_query(mac: str, interval: float, timeout: float, include_raw: bool, args) -> int:
"""
Continuously query the battery and publish results.
Args:
mac: BLE MAC address.
interval: Seconds between queries.
timeout: Seconds to wait for each notification.
include_raw: Whether to include raw_hex.
args: Parsed command line arguments (for MQTT settings).
Returns:
Exit code (0 on success, >0 on error).
"""
device = await BleakScanner.find_device_by_address(mac, timeout=8.0)
if not device:
status = Status(status="error", timestamp=utc_now_iso(), message=f"Device {mac} not found")
publish_stdout(status)
if args.publish == "mosquitto":
publish_offline(args)
return 2
async with BleakClient(device) as client:
char = await find_characteristic(client, CHAR_UUID)
if not char:
status = Status(status="error", timestamp=utc_now_iso(), message="Characteristic not found")
publish_stdout(status)
if args.publish == "mosquitto":
publish_offline(args)
return 3
result: Optional[Status] = None
def handler(sender, data):
nonlocal result
result = decode_status(data, include_raw=include_raw)
await client.start_notify(char, handler)
# Warm-up: first query with longer timeout, result is discarded
result = None
await client.write_gatt_char(char, CMD_REQUEST_STATUS, response=False)
for _ in range(50): # 5 seconds wait (50 * 0.1s)
if result is not None:
break
await asyncio.sleep(0.1)
try:
while True:
result = None
await client.write_gatt_char(char, CMD_REQUEST_STATUS, response=False)
# Simple polling loop
for _ in range(int(timeout * 10)):
if result is not None:
break
await asyncio.sleep(0.1)
if result:
status = result
else:
status = Status(status="error", timestamp=utc_now_iso(), message="Timeout")
publish_stdout(status)
if args.publish == "mosquitto":
publish_mosquitto(status, args)
if args.once:
break
await asyncio.sleep(interval)
finally:
await client.stop_notify(char)
if args.publish == "mosquitto":
publish_offline(args)
return 0
# ---------- Main entry point ----------
async def async_main() -> int:
"""Async main routine."""
parser = build_parser()
args = parser.parse_args()
setup_logging(args.verbose)
if args.publish == "mosquitto" and which("mosquitto_pub") is None:
LOGGER.error("mosquitto_pub not found – install mosquitto-clients")
return 4
if args.mode == "once":
status = await read_once(args.mac, args.timeout, args.raw)
publish_stdout(status)
if args.publish == "mosquitto":
publish_mosquitto(status, args)
publish_offline(args)
return 0 if status.status == "ok" else 1
return await cyclic_query(args.mac, args.interval, args.timeout, args.raw, args)
def main() -> int:
"""Synchronous entry point."""
try:
return asyncio.run(async_main())
except KeyboardInterrupt:
print("\nAbbruch durch Benutzer", file=sys.stderr)
return 0
if __name__ == "__main__":
raise SystemExit(main())
, speicher es z.b. als “accubox-ble.py” und dann musst du es noch ausführbar machen mittels
chmod +x accubox-ble.py
und testen mittels
./accubox-ble.py --mac 58:b6:4f:49:9c:f1
. Die MAC musst du mit deiner ersetzen (kurzes Stossgebet ) und ENTER drücken, es sollte was in dieser Art zusehen sein
{"status": "ok", "timestamp": "2026-06-01T17:27:52.856708+02:00", "capacity_ah": 182, "soc_percent": 91, "rest_time": "143:32", "voltage_v": 13.3, "current_a": -0.29, "power_w": -3.9, "mode": "discharging", "raw_hex": null, "message": null}
Meine AccuBox ist eine S200 von 2023. Ob das Protokoll bei deiner S300 identisch ist, weiß ich nicht – aber ein Test wäre es ja mal wert…
Soweit erstmal. Vielleicht haut es ja hin, würde mich freuen.
Liebe Grüße,
Erik
P.S.: Es ist einiges an Code dazu gekommen durch MQQT , ich denke das lässt sich auch mittels Node-Red regeln aber vielleicht ist es ja so auch praktisch, die leichtere Version ohne MQQT in dem anderen Topic sollte jedenfalls genauso funktionieren.
Hi und danke erstmal.
Habe lange am Auto gebaut und die Software eigendlich links liegen gelassen, da ich von dem nicht viel verstehe und eigendlich hoffte irgendwer anderes macht den Versuch vor mir.
War wohl leider nicht so.
Bin inzwischen mit dem Fahrzeug unterwegs das so um die 75% fertig ist, was mir für jetzt gerade reicht.
Bin gerade in der Toscana auf einen Stellplatz und habe Bock was an der Software zu machen um die Akkubox auszulesen.
Ich scheitere aber bereits an der Installation des Bleak.
Servus
Klaus
pi@pekaway:~ $ pip install --user bleak
error: externally-managed-environment
× This environment is externally managed
╰─> To install Python packages system-wide, try apt install
python3-xyz, where xyz is the package you are trying to
install.
If you wish to install a non-Debian-packaged Python package,
create a virtual environment using python3 -m venv path/to/venv.
Then use path/to/venv/bin/python and path/to/venv/bin/pip. Make
sure you have python3-full installed.
For more information visit http://rptl.io/venv
note: If you believe this is a mistake, please contact your Python installation or OS distribution provider. You can override this, at the risk of breaking your Python installation or OS, by passing --break-system-packages.
hint: See PEP 668 for the detailed specification.
pi@pekaway:~ $
Hallo Klaus
Alle die sich da auskennen sind sicher in Urlaub 
Aber schau mal hier rein, da waren die gleichen Probleme, und 2 Lösungen
Gruß Arno
Hallo zusammen, @Klaus dass ist das klassisches pip Hausverbot, hätte ich dran denken können 
aber wie oben erwähnt würde ich eh erstmal einen Laptop zum testen nehmen bevor du es auf deinem VanPi testest/nutzt, sollte das ein Windows Laptop sein dann wäre pip denke ich die Variante auf OS X ist es dann brew aber auf Linux gehen viele Paket-Manager wie auch auf dem Pi der apt nutzt. Du kannst es dort mit einem
sudo apt install python3-bleak
installieren. So findest du auch infos zu dem Paket
sudo apt info python3-bleak
. Solltest du es installiert haben dann kannst du deine MAC auch so rausfinden
python3 -c “import asyncio; from bleak import BleakScanner; devs = asyncio.run(BleakScanner.discover(timeout=5.0)); [print(f’{d.address} - {d.name}') for d in devs if d.name]”
das sollte dir alle MACs anzeigen (meine ist mit einem Datum im Namen zufinden 58:B6:4F:49:9C:F1 - 23-04-30-028 ) u.a. auch die deiner AccuBox wo du dann das --mac Flag im Skript verwenden kannst.
Ich habe einige Updates an dem Skript gemacht was mittlerweile günstiger für mich ist und auch ein bisschen stabiler, vielleicht auch besser für dich auf dem VanPi zu gebrauchen, kein MQQT mehr drinne aber das sollte bei dir auch über Node Red über die RAM Disk gut gehen ??? Hier mal das neue Skript
#!/usr/bin/env python3
“”"
AccuBox BLE JSON Reader - v4.1
Schlanker BLE-Treiber fuer die Ective AccuBox LiFePO4 Batterie.
Alle Statistiken und Validierung werden von cockpit_api.py verwaltet
(Single Source of Truth Prinzip).
Aenderungen gegenueber v4.0:
decode_status(): Tabs/Spaces-Mix beseitigt (IndentationError behoben)
decode_status(): temp_c Zuweisung aus falschem if-Block befreit
decode_status(): result-Dictionary und return auf korrekte Ebene gesetzt
decode_status(): Sekundaere Modus-Erkennung via dekodierter Spannung
(LiFePO4 Absorptions-/Float-Phase bei voltage_v >= 14.0V wird korrekt
als Laden erkannt, auch wenn BMS voltage_raw <= 200 meldet)
decode_status(): time_label Logik vereinfacht und korrigiert
Gesamte Datei einheitlich mit 4 Spaces eingerueckt (kein Tab-Mix mehr)
Bekannte BMS-Eigenheiten (Ective AccuBox):
BMS cached Antworten intern ca. 5 Sekunden
→ ca. 12-15% Duplikate bei 5s Abfrageintervall
→ Duplicate Detection erfolgt in cockpit_api.py
BMS SOC kann > 100% anzeigen wenn nicht kalibriert
→ Plausibilitaetspruefung erfolgt in cockpit_api.py
BMS wechselt voltage_raw Bereich bei sehr kleinem Ladestrom
→ Sekundaere Modus-Erkennung via voltage_v in decode_status()
Output-Format (/dev/shm/accubox.json):
{
“status”: “ok”,
“timestamp”: “2026-06-18T15:30:15.123456+02:00”,
“capacity_ah”: 198,
“soc_percent”: 99,
“voltage_v”: 14.6,
“current_a”: 0.51,
“power_w”: 7.5,
“temp_c”: 22.1,
“mode”: “charging”,
“time_remaining”: “01:44”,
“time_label”: “time_to_full”,
“meta”: {
“sequence”: 1234,
“ble_connected”: true,
“query_duration_ms”: 342
}
}
Protokoll (21 Bytes, Little-Endian):
[0:4] Header (1c031000)
[4:6] Kapazitaet (Ah)
[6:8] Ladezustand (%)
[8:10] Verbleibende Stunden
[10:12] Verbleibende Minuten
[12:14] Ladestrom Raw (signed int16)
[14:16] Temperatur Raw (uint16)
[16:18] Spannung Raw (uint16)
[18:20] Entladestrom Raw (signed int16)
[20] Checksum/Padding
Modus-Erkennung (zwei Stufen):
Primaer: voltage_raw > 200 → Lademodus
voltage_raw > 200: voltage / 47.3, charge_current / 94.0
voltage_raw <= 200: voltage / 10.0, discharge_current / 7965.0
Sekundaer: voltage_v >= 14.0V → ebenfalls Laden
(BMS kippt bei Float-Phase in Entladebereich, obwohl noch Laden)
Location: /home/walter/server/api/devices/accubox_ble.py
Author: ummeegge + AI assistance
License: MIT
Version: 4.1
“”"
import asyncio
import json
import argparse
import sys
import os
import signal
import time
from datetime import datetime, timezone
from typing import Optional
from bleak import BleakScanner, BleakClient, BleakError
============================================================
KONFIGURATION
============================================================
DEFAULT_MAC = “58:b6:4f:49:9c:f1”
CHAR_UUID = “0000ffe1-0000-1000-8000-00805f9b34fb”
CMD_REQUEST_STATUS = bytes.fromhex(“1c030400000846b1”)
Skalierungsfaktoren (empirisch verifiziert gegen Ective App + Wireshark)
VOLT_SCALING_CHARGE = 47.3
VOLT_SCALING_DISCHARGE = 10.0
CURRENT_SCALING_CHARGE = 94.0
CURRENT_SCALING_DISCHARGE = 7965.0
voltage_raw Schwellwert fuer primaere Modus-Erkennung
VOLTAGE_RAW_CHARGE_THRESHOLD = 200
Sekundaerer Schwellwert: Ab dieser dekodierten Spannung gilt “charging”,
auch wenn voltage_raw <= 200 (BMS Float-/Absorptionsphase)
VOLTAGE_SECONDARY_CHARGE_THRESHOLD = 14.0
Temperatur-Skalierung (empirisch kalibriert: 23°C Umgebung ≈ temp_raw 184)
TEMP_SCALING = 8.0
Reconnect-Timing (Exponential Backoff)
INITIAL_RETRY_DELAY = 5.0
MAX_RETRY_DELAY = 60.0
BLE Warm-up Timing nach Connect
WARMUP_DELAY_SEC = 0.5
WARMUP_TIMEOUT_SEC = 2.0
Globales Flag fuer Graceful Shutdown (wird durch Signal-Handler gesetzt)
_shutdown_requested = False
============================================================
SIGNAL HANDLER
============================================================
def _install_signal_handlers() → None:
“”"
Installiert Signal-Handler fuer SIGTERM und SIGINT.
Setzt das globale _shutdown_requested Flag auf True,
damit der asyncio Event-Loop in cyclic_query() sauber
aus seiner while-Schleife aussteigen kann.
Verhindert dadurch den SIGKILL durch systemd nach
TimeoutStopSec (aktuell 10s).
"""
def _handler(signum: int, frame) -> None:
global _shutdown_requested
signame = signal.Signals(signum).name
_log_info(f"Signal {signame} empfangen - beende sauber...")
_shutdown_requested = True
signal.signal(signal.SIGTERM, _handler)
signal.signal(signal.SIGINT, _handler)
============================================================
HILFSFUNKTIONEN
============================================================
def now_iso() → str:
“”“Gibt die aktuelle Uhrzeit als timezone-aware ISO 8601 String zurueck.”“”
return datetime.now(timezone.utc).astimezone().isoformat()
def _log_info(message: str) → None:
“”“Gibt eine Info-Nachricht als JSON-Zeile auf stderr aus.”“”
print(
json.dumps({
“status”: “info”,
“message”: message,
“timestamp”: now_iso(),
}),
file=sys.stderr,
flush=True,
)
def _log_warning(message: str) → None:
“”“Gibt eine Warn-Nachricht als JSON-Zeile auf stderr aus.”“”
print(
json.dumps({
“status”: “warning”,
“message”: message,
“timestamp”: now_iso(),
}),
file=sys.stderr,
flush=True,
)
def _atomic_write(filepath: str, content: str) → None:
“”"
Schreibt content atomar in filepath.
Nutzt eine temporaere Datei + os.replace() damit der
Leser (cockpit_api.py) nie eine halbfertige Datei sieht.
Besonders wichtig auf der RAM-Disk /dev/shm/.
Args:
filepath: Zielpfad (z.B. /dev/shm/accubox.json)
content: JSON-String der geschrieben werden soll
"""
tmp = filepath + ".tmp"
try:
with open(tmp, "w") as f:
f.write(content)
os.replace(tmp, filepath)
except OSError as e:
_log_warning(f"Atomic write fehlgeschlagen ({filepath}): {e}")
============================================================
PROTOKOLL-DEKODIERUNG
============================================================
def decode_status(data: bytes, include_raw: bool = False) → dict:
“”"
Dekodiert die 21-Byte BMS-Notification in ein lesbares Dictionary.
Protokoll-Details:
Das BMS der Ective AccuBox sendet immer exakt 21 Bytes
im Little-Endian Format als BLE Notification.
Modus-Erkennung (zwei Stufen):
Stufe 1 - voltage_raw Schwellwert:
voltage_raw > 200: Lademodus
voltage = voltage_raw / 47.3
current = abs(charge_current_raw) / 94.0 (positiv)
voltage_raw <= 200: Entlademodus (Kandidat)
voltage = voltage_raw / 10.0
current = -abs(discharge_current_raw) / 7965.0 (negativ)
Stufe 2 - dekodierte Spannung (Korrektur fuer Float-Phase):
Das BMS kann bei sehr kleinem Ladestrom (Absorptions-/
Float-Phase) in den voltage_raw <= 200 Bereich kippen,
obwohl die Batterie noch geladen wird.
Wenn voltage_v >= 14.0V: trotzdem "charging"
→ Strom wird umgekehrt (positiv)
Wichtige Protokoll-Notiz:
data[13] ist KEIN Multiplexer-Byte! Es ist das High-Byte
des Charge Current Raw-Werts (data[12:14] = signed int16).
Fruehere Versionen haben das faelschlicherweise als MUX
interpretiert (Bug in v2.x).
Args:
data: 21 Bytes BMS-Notification
include_raw: Wenn True wird raw_hex zum Dictionary hinzugefuegt
Returns:
Dictionary mit dekodierten Werten (status="ok")
oder Fehler-Dictionary (status="error")
"""
if len(data) != 21:
return {
"status": "error",
"timestamp": now_iso(),
"message": f"Unerwartete Paketlaenge: {len(data)} (erwartet: 21)",
}
# --- Byte-Extraktion ---
capacity_ah = int.from_bytes(data[4:6], "little")
soc_percent = int.from_bytes(data[6:8], "little")
rest_hours = int.from_bytes(data[8:10], "little")
rest_minutes = int.from_bytes(data[10:12], "little")
temp_raw = int.from_bytes(data[14:16], "little")
voltage_raw = int.from_bytes(data[16:18], "little")
current_charge_raw = int.from_bytes(data[12:14], "little", signed=True)
current_discharge_raw = int.from_bytes(data[18:20], "little", signed=True)
# --- Modus-Erkennung und Skalierung (Stufe 1) ---
if voltage_raw > VOLTAGE_RAW_CHARGE_THRESHOLD:
# Primaerer Lademodus: BMS meldet hohen voltage_raw
mode = "charging"
voltage_v = round(voltage_raw / VOLT_SCALING_CHARGE, 2)
current_a = round(abs(current_charge_raw) / CURRENT_SCALING_CHARGE, 2)
else:
# Entlademodus-Kandidat: voltage_raw klein
voltage_v = round(voltage_raw / VOLT_SCALING_DISCHARGE, 2)
current_a = -round(abs(current_discharge_raw) / CURRENT_SCALING_DISCHARGE, 2)
# --- Modus-Korrektur (Stufe 2) ---
# Bei Float-/Absorptionsphase kann voltage_v >= 14.0V sein,
# obwohl voltage_raw <= 200 (BMS-interne Umschaltung).
# In diesem Fall behandeln wir es weiterhin als Laden.
if voltage_v >= VOLTAGE_SECONDARY_CHARGE_THRESHOLD:
mode = "charging"
current_a = abs(current_a) # Strom umkehren: negativ -> positiv
else:
mode = "discharging"
# --- Leistung (Vorzeichen folgt dem Strom) ---
power_w = round(abs(current_a) * voltage_v, 1)
if current_a < 0:
power_w = -power_w
# --- Temperatur ---
# temp_raw >= 65000 deutet auf einen BLE-Glitch hin (uint16 Overflow)
temp_c = round(temp_raw / TEMP_SCALING, 1) if temp_raw < 65000 else None
# --- Restzeit und Zeit-Label ---
rest_time = f"{rest_hours:02d}:{rest_minutes:02d}"
if mode == "charging":
time_label = "time_to_full"
elif mode == "discharging" and current_a < 0:
time_label = "time_to_empty"
else:
# Randfall: Strom = 0.0 (Batterie voll oder idle)
time_label = "time_remaining"
# --- Ergebnis zusammenbauen ---
result = {
"status": "ok",
"timestamp": now_iso(),
"capacity_ah": capacity_ah,
"soc_percent": soc_percent,
"voltage_v": voltage_v,
"current_a": current_a,
"power_w": power_w,
"temp_c": temp_c,
"mode": mode,
"time_remaining": rest_time,
"time_label": time_label,
}
if include_raw:
result["raw_hex"] = data.hex()
return result
============================================================
BLE HILFSFUNKTIONEN
============================================================
async def _find_characteristic(client: BleakClient, uuid: str):
“”"
Sucht eine GATT-Charakteristik anhand ihrer UUID.
Iteriert ueber alle Services und Charakteristiken des
verbundenen Geraets. UUID-Vergleich ist case-insensitiv.
Args:
client: Verbundener BleakClient
uuid: UUID der gesuchten Charakteristik (beliebige Schreibweise)
Returns:
BleakGATTCharacteristic oder None wenn nicht gefunden
"""
for service in client.services:
for char in service.characteristics:
if char.uuid.lower() == uuid.lower():
return char
return None
async def _warmup_bms(
client: BleakClient,
char,
queue: asyncio.Queue,
) → bool:
“”"
Fuehrt eine Warm-up-Abfrage nach dem BLE-Connect durch.
Das BMS der AccuBox antwortet haeufig nicht auf die allererste
Anfrage nach einer neuen BLE-Verbindung (interner Initialisierungs-
delay). Diese Funktion sendet eine Dummy-Anfrage und absorbiert
den Timeout, damit der eigentliche Datenfluss sauber startet.
Args:
client: Verbundener BleakClient
char: Ziel-Charakteristik fuer den Write
queue: asyncio.Queue die Notifications empfaengt
Returns:
True wenn BMS auf Warm-up geantwortet hat
False bei Timeout (normal, kein Fehler)
"""
# Eventuelle alte Notifications aus Queue raeumen
while not queue.empty():
queue.get_nowait()
await asyncio.sleep(WARMUP_DELAY_SEC)
try:
await client.write_gatt_char(char, CMD_REQUEST_STATUS, response=False)
await asyncio.wait_for(queue.get(), timeout=WARMUP_TIMEOUT_SEC)
return True
except asyncio.TimeoutError:
return False
============================================================
EINMALIGE ABFRAGE
============================================================
async def read_once(
mac: str,
timeout: float,
include_raw: bool,
) → None:
“”"
Fuehrt eine einzelne BLE-Abfrage durch und gibt das Ergebnis aus.
Verbindet mit dem BLE-Geraet, sendet einen Status-Request,
wartet auf die Notification, gibt das dekodierte JSON auf
stdout aus und trennt die Verbindung.
Geeignet fuer manuelle Tests und Debugging:
python3 accubox_ble.py --mode once
python3 accubox_ble.py --mode once --mac AA:BB:CC:DD:EE:FF
Args:
mac: BLE MAC-Adresse des Geraets
timeout: Max. Wartezeit in Sekunden fuer die Notification
include_raw: Wenn True wird raw_hex in der Ausgabe enthalten
"""
device = await BleakScanner.find_device_by_address(mac, timeout=8.0)
if not device:
print(json.dumps({
"status": "error",
"message": f"Geraet {mac} nicht gefunden",
}))
return
async with BleakClient(device) as client:
target_char = await _find_characteristic(client, CHAR_UUID)
if not target_char:
print(json.dumps({
"status": "error",
"message": "Charakteristik nicht gefunden",
}))
return
result_queue: asyncio.Queue = asyncio.Queue()
def handler(sender, data: bytearray) -> None:
result_queue.put_nowait(decode_status(bytes(data), include_raw))
await client.start_notify(target_char, handler)
# Warm-up: erste Antwort des BMS absorbieren
await _warmup_bms(client, target_char, result_queue)
# Queue leeren (Warm-up Antwort verwerfen)
while not result_queue.empty():
result_queue.get_nowait()
# Echte Abfrage senden
await client.write_gatt_char(
target_char, CMD_REQUEST_STATUS, response=False
)
try:
result = await asyncio.wait_for(result_queue.get(), timeout=timeout)
print(json.dumps(result, ensure_ascii=False, indent=2))
except asyncio.TimeoutError:
print(json.dumps({
"status": "error",
"message": "Timeout: Keine Antwort vom BMS",
}))
finally:
await client.stop_notify(target_char)
============================================================
ZYKLISCHE ABFRAGE (PRODUCTION)
============================================================
async def cyclic_query(
mac: str,
interval_sec: float,
timeout: float,
include_raw: bool,
outfile: Optional[str] = None,
) → None:
“”"
Fuehrt zyklische BLE-Abfragen im Dauer-Monitoring-Modus durch.
Laeuft als Endlos-Schleife bis _shutdown_requested gesetzt wird.
Schreibt jedes valide Ergebnis auf stdout (fuer journald) und
optional atomar in eine JSON-Datei (fuer cockpit_api.py).
Verbindungsverhalten:
- Bei Verbindungsverlust: Exponential Backoff (5s bis 60s)
- Bei erfolgreichem Reconnect: Backoff zuruecksetzen auf 5s
- BLE Warm-up nach jedem Connect
Shutdown-Verhalten:
- SIGTERM/SIGINT setzt _shutdown_requested = True
- Naechste Loop-Iteration bricht sauber ab
- BLE-Verbindung wird ordentlich getrennt
- Kein SIGKILL durch systemd noetig (innerhalb TimeoutStopSec)
Args:
mac: BLE MAC-Adresse des Geraets
interval_sec: Abfrage-Intervall in Sekunden (5s empfohlen)
timeout: Max. Wartezeit in Sekunden fuer eine Notification
include_raw: Wenn True wird raw_hex in JSON ausgegeben
outfile: Pfad fuer atomaren JSON-Write, z.B. /dev/shm/accubox.json
"""
global _shutdown_requested
retry_delay = INITIAL_RETRY_DELAY
sequence = 0
while not _shutdown_requested:
_log_info(f"Scanne nach {mac}...")
# --- Geraet suchen ---
device = await BleakScanner.find_device_by_address(mac, timeout=8.0)
if not device:
_log_warning(f"Geraet nicht gefunden. Retry in {retry_delay:.0f}s")
await asyncio.sleep(retry_delay)
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
continue
# --- Verbinden und Abfrage-Schleife ---
try:
async with BleakClient(device) as client:
target_char = await _find_characteristic(client, CHAR_UUID)
if not target_char:
raise BleakError("Charakteristik nicht gefunden")
result_queue: asyncio.Queue = asyncio.Queue()
def handler(sender, data: bytearray) -> None:
result_queue.put_nowait(
decode_status(bytes(data), include_raw)
)
await client.start_notify(target_char, handler)
# Warm-up
warmup_ok = await _warmup_bms(client, target_char, result_queue)
_log_info(
f"Verbunden. Abfrage alle {interval_sec}s. "
f"BMS {'bereit' if warmup_ok else 'bereit (Warm-up Timeout absorbiert)'}"
)
# Erfolgreiche Verbindung: Backoff zuruecksetzen
retry_delay = INITIAL_RETRY_DELAY
try:
while not _shutdown_requested:
# Alte Notifications aus Queue raeumen
while not result_queue.empty():
result_queue.get_nowait()
# Abfrage senden und Zeit messen
query_start = time.monotonic()
await client.write_gatt_char(
target_char,
CMD_REQUEST_STATUS,
response=False,
)
try:
result = await asyncio.wait_for(
result_queue.get(), timeout=timeout
)
if result.get("status") == "ok":
sequence += 1
result["meta"] = {
"sequence": sequence,
"ble_connected": bool(client.is_connected),
"query_duration_ms": round(
(time.monotonic() - query_start) * 1000
),
}
json_out = json.dumps(result, ensure_ascii=False)
print(json_out, flush=True)
if outfile:
_atomic_write(outfile, json_out)
else:
# Fehler-Response (z.B. falsches Paket) durchreichen
print(
json.dumps(result, ensure_ascii=False),
flush=True,
)
except asyncio.TimeoutError:
_log_warning("BLE Notification Timeout - BMS reagiert nicht")
await asyncio.sleep(interval_sec)
except asyncio.CancelledError:
_log_info("Abfrage-Schleife abgebrochen")
finally:
# BLE-Verbindung sauber abmelden
try:
await client.stop_notify(target_char)
except Exception:
pass
_log_info("BLE-Verbindung sauber getrennt")
except (BleakError, asyncio.TimeoutError, OSError) as e:
_log_warning(
f"Verbindungsfehler: {type(e).__name__}: {e}. "
f"Retry in {retry_delay:.0f}s"
)
await asyncio.sleep(retry_delay)
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
_log_info("Shutdown abgeschlossen.")
============================================================
KOMMANDOZEILEN-INTERFACE
============================================================
def main() → None:
“”"
Einstiegspunkt des AccuBox BLE Readers.
Verarbeitet Kommandozeilen-Argumente und startet den
passenden Abfragemodus.
Beispiele:
Einmalige Abfrage (Debugging):
python3 accubox_ble.py --mode once
Einmalige Abfrage mit Raw-Hex (Protokoll-Analyse):
python3 accubox_ble.py --mode once
Production (via systemd, ohne Raw-Hex):
python3 accubox_ble.py \\
--mode cyclic \\
--interval 5 \\
--no-raw \\
--outfile /dev/shm/accubox.json
Anderes Geraet:
python3 accubox_ble.py --mode once --mac AA:BB:CC:DD:EE:FF
"""
parser = argparse.ArgumentParser(
description="AccuBox BLE JSON Reader v4.1",
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
)
parser.add_argument(
"--mode",
choices=["once", "cyclic"],
default="once",
help="once: einmalige Abfrage | cyclic: Dauer-Monitoring",
)
parser.add_argument(
"--interval",
type=float,
default=5.0,
help="Abfrage-Intervall in Sekunden (5s empfohlen fuer Raspberry Pi)",
)
parser.add_argument(
"--timeout",
type=float,
default=3.0,
help="Max. Wartezeit in Sekunden fuer BLE-Notification",
)
parser.add_argument(
"--mac",
type=str,
default=DEFAULT_MAC,
help="BLE MAC-Adresse der AccuBox",
)
parser.add_argument(
"--no-raw",
action="store_true",
help="raw_hex aus JSON-Ausgabe weglassen (empfohlen fuer Production)",
)
parser.add_argument(
"--outfile",
type=str,
default=None,
help="JSON atomar in diese Datei schreiben (z.B. /dev/shm/accubox.json)",
)
args = parser.parse_args()
include_raw = not args.no_raw
# Signal-Handler installieren bevor asyncio.run() gestartet wird
_install_signal_handlers()
if args.mode == "once":
asyncio.run(read_once(args.mac, args.timeout, include_raw))
else:
try:
asyncio.run(
cyclic_query(
args.mac,
args.interval,
args.timeout,
include_raw,
args.outfile,
)
)
except KeyboardInterrupt:
_log_info("Keyboard Interrupt - auf Wiedersehen!")
sys.exit(0)
if name == “main”:
main()
. Einfach Ausgabe sieht hier so aus
╰─➤ server/api/devices/accubox_ble.py
{
“status”: “ok”,
“timestamp”: “2026-07-17T12:28:01.409469+02:00”,
“capacity_ah”: 120,
“soc_percent”: 60,
“voltage_v”: 13.72,
“current_a”: 38.13,
“power_w”: 523.1,
“temp_c”: 23.8,
“mode”: “charging”,
“time_remaining”: “02:06”,
“time_label”: “time_to_full”,
“raw_hex”: “1c03100078003c0002000600000ebe008902053f8d”
} .
Du kannst mit dem --output Flag in die RAM Disk schreiben e.g. –outfile /dev/shm/accubox.json Vielleicht ist es erstmal das bessere so ?!?!
Kannst ja erstmal schauen ob du es zum laufen bekommst.
Herzliche Grüße,
Erik
Hi Erik
bin zurück aus dem Urlaub und werde mich diese Tage an deinem Script versuchen.
Danke
Klaus
Grüß dich Klaus,
ja mach das gerne! Ich habe allerdings gesehen, dass die
Forensoftware den Code “aufgehübscht” hat (Smart Punctuation)
und manche Sonderzeichen geändert hat. Leider kann ich meinen
alten Post auch nicht mehr editieren.
Ich habe dir deshalb einen Gist auf GitHub angelegt, wo alles
sauber drinsteht. Da ich hier als neuer User noch keine Links
posten darf: einfach auf gist.github.com nach “ummeegge” suchen
und den Gist “AccuBox BLE JSON Reader” öffnen. Falls du ihn
nicht findest, schreib mir kurz eine PN, dann schicke
ich dir den Link direkt.
Bei mir geht’s morgen in den Urlaub — der Balkan ruft — also
bin ich wahrscheinlich auch öfters mal off.
Beste Grüße