Titan 2,4 kW Akku Pylontech BMS auslesen

[antaril]

Hi Leute,

Hat schon jemand es geschafft die Infos vom Akku via Skript auszulesen? Ich bekomm zwar antworten vom BMS aber irgendwie passt da was nicht. Evt hat das ja schon wer gemacht.

Greets Antaril

[antaril]

ALso ich habe es geschafft den akku auszulesen und die werte zu interpretieren: Also der pythoncode ist so:

#!/usr/bin/env python3

from datetime import datetime
import os
import time
import serial

PORT = "/dev/ttyUSB0"
BAUDRATE = 9600

FIRST_ADDRESS = 1
LAST_ADDRESS = 16

RESPONSE_TIMEOUT = 1.5
DELAY_BETWEEN_REQUESTS = 0.20
INTERVAL = 5.0

OUTPUT_FILE = "/var/www/html/akku2.txt"

HISTORY_DIR = "/var/lib/akku2"
HISTORY_FILE = os.path.join(HISTORY_DIR, "capacity-history.txt")

============================================================

PYLONTECH / PACE

============================================================

def checksum(body):
total = sum(body.encode("ascii"))
return (-total) & 0xFFFF

def make_request(address):
body = f"20{address:02X}4642E002{address:02X}"
chk = checksum(body)
return f"~{body}{chk:04X}\r".encode("ascii")

def read_response(ser, address):
ser.reset_input_buffer()
ser.write(make_request(address))
ser.flush()

deadline = time.time() + RESPONSE_TIMEOUT
data = bytearray()

while time.time() < deadline:
if chunk := ser.read(256):
data.extend(chunk)
if b"\r" in data:
break
time.sleep(0.01)

if not data:
return None

try:
text = data.decode("ascii", errors="ignore")
start, end = text.find("~"), text.find("\r")
if start < 0 or end < 0:
return None

frame = text[start:end]
if len(frame) < 20:
  return None

body = frame[1:]
if int(body[-4:], 16) != checksum(body[:-4]):
  return None

return body[12:]

except Exception:
return None

def parse_response(info):
try:
if not info or len(info) < 10:
return None

info_flag = int(info[0:2], 16)
pack_id = int(info[2:4], 16)
cell_count = int(info[4:6], 16)
pos = 6

cells = []
for _ in range(cell_count):
  if pos + 4 > len(info):
    return None
  cells.append(int(info[pos : pos + 4], 16) / 1000.0)
  pos += 4

if pos + 2 > len(info):
  return None
temp_count = int(info[pos : pos + 2], 16)
pos += 2

temps = []
for _ in range(temp_count):
  if pos + 4 > len(info):
    return None
  raw = int(info[pos : pos + 4], 16)
  temps.append((raw - 2731) / 10.0)
  pos += 4

if pos + 4 > len(info):
  return None
current_raw = int(info[pos : pos + 4], 16)
if current_raw >= 0x8000:
  current_raw -= 0x10000
current = current_raw / 10.0
pos += 4

if pos + 4 > len(info):
  return None
voltage = int(info[pos : pos + 4], 16) / 1000.0
pos += 4

remaining_hex = info[pos:]
remaining_raw = total_raw = user_defined = cycles = None

if len(remaining_hex) >= 14:
  try:
    remaining_raw = int(remaining_hex[0:4], 16)
    user_defined = int(remaining_hex[4:6], 16)
    total_raw = int(remaining_hex[6:10], 16)
    cycles = int(remaining_hex[10:14], 16)
  except Exception:
    pass

remaining_ah = remaining_raw / 100.0 if remaining_raw is not None else None
total_ah = total_raw / 100.0 if total_raw is not None else None

soc = None
if (
    remaining_raw is not None
    and total_raw is not None
    and total_raw > 0
    and remaining_raw <= total_raw
):
  soc = max(0.0, min(100.0, (remaining_raw / total_raw) * 100.0))

active = len(cells) > 0 and max(cells) > 1.0 and voltage > 10.0

return {
    "active": active,
    "info_flag": info_flag,
    "pack_id": pack_id,
    "cells": cells,
    "temps": temps,
    "current": current,
    "voltage": voltage,
    "power": voltage * current,
    "remaining_hex": remaining_hex,
    "remaining_raw": remaining_raw,
    "total_raw": total_raw,
    "remaining_ah": remaining_ah,
    "total_ah": total_ah,
    "remaining_kwh": (
        remaining_ah * voltage / 1000.0 if remaining_ah else None
    ),
    "total_kwh": total_ah * voltage / 1000.0 if total_ah else None,
    "user_defined": user_defined,
    "cycles": cycles,
    "soc": soc,
    "raw_info": info,
}

except Exception:
return None

============================================================

KAPAZITÄTSHISTORIE

============================================================

def read_max_capacity():
try:
if not os.path.exists(HISTORY_FILE):
return None

maximum = None
with open(HISTORY_FILE, "r") as f:
  for line in f:
    if line := line.strip():
      parts = line.split(";")
      if len(parts) >= 3:
        try:
          val = float(parts[2])
          if maximum is None or val > maximum:
            maximum = val
        except ValueError:
          continue
return maximum

except Exception:
return None

def update_capacity_history(address, data):
if data["total_ah"] is None:
return None

try:
os.makedirs(HISTORY_DIR, exist_ok=True)
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
total_ah = data["total_ah"]
cycles = data["cycles"] if data["cycles"] is not None else -1

last_value = None
if os.path.exists(HISTORY_FILE):
  try:
    with open(HISTORY_FILE, "r") as f:
      if lines := f.readlines():
        parts = lines[-1].strip().split(";")
        if len(parts) >= 3:
          last_value = float(parts[2])
  except Exception:
    pass

if last_value is None or abs(total_ah - last_value) >= 0.01:
  with open(HISTORY_FILE, "a") as f:
    f.write(f"{timestamp};{address};{total_ah:.2f};{cycles}\n")

max_capacity = read_max_capacity()
if max_capacity is None or total_ah > max_capacity:
  max_capacity = total_ah

return max_capacity

except Exception:
return None

============================================================

AKKU FINDEN

============================================================

def find_active_battery(ser):
for address in range(FIRST_ADDRESS, LAST_ADDRESS + 1):
info = read_response(ser, address)
if info and (data := parse_response(info)) and data["active"]:
return address, data
time.sleep(DELAY_BETWEEN_REQUESTS)
return None, None

============================================================

TXT-DATEI

============================================================

def write_file(address, data, max_capacity):
cells, temps = data["cells"], data["temps"]
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

lines = [
f"timestamp={timestamp}",
f"address={address}",
f"voltage={data['voltage']:.3f}",
f"current={data['current']:.1f}",
f"power={data['power']:.1f}",
f"cell_count={len(cells)}",
]

if cells:
cell_min, cell_max = min(cells), max(cells)
cell_delta = cell_max - cell_min
lines.extend([
f"cell_min={cell_min:.3f}",
f"cell_max={cell_max:.3f}",
f"cell_delta={cell_delta:.3f}",
f"cell_delta_mv={round(cell_delta * 1000)}",
])
for i, val in enumerate(cells, 1):
lines.append(f"cell{i}={val:.3f}")

lines.append(f"temp_count={len(temps)}")
if temps:
lines.extend([
f"temp_min={min(temps):.1f}",
f"temp_max={max(temps):.1f}",
f"temp_avg={sum(temps) / len(temps):.1f}",
])
for i, val in enumerate(temps, 1):
lines.append(f"temp{i}={val:.1f}")

for key in ["remaining_raw", "total_raw", "user_defined", "cycles"]:
if data[key] is not None:
lines.append(f"{key}={data[key]}")

for key in ["remaining_ah", "total_ah"]:
if data[key] is not None:
lines.append(f"{key}={data[key]:.2f}")

for key in ["remaining_kwh", "total_kwh"]:
if data[key] is not None:
lines.append(f"{key}={data[key]:.3f}")

if data["soc"] is not None:
lines.extend(
[f"soc={data['soc']:.1f}", "soc_source=bms_capacity_ratio"]
)
else:
lines.extend(["soc=unknown", "soc_source=unknown"])

soh = None
if max_capacity and max_capacity > 0 and data["total_ah"] is not None:
soh = max(0.0, min(100.0, (data["total_ah"] / max_capacity) * 100.0))

if soh is not None:
lines.extend([f"soh={soh:.1f}", "soh_source=observed_max_capacity"])
else:
lines.extend(["soh=unknown", "soh_source=waiting_for_reference"])

if max_capacity is not None:
lines.append(f"max_observed_capacity_ah={max_capacity:.2f}")

lines.extend(
[f"remaining_hex={data['remaining_hex']}", f"raw_info={data['raw_info']}"]
)

content = "\n".join(lines) + "\n"
tmp_file = OUTPUT_FILE + ".tmp"

try:
with open(tmp_file, "w") as f:
f.write(content)
os.replace(tmp_file, OUTPUT_FILE)
except Exception:
if os.path.exists(tmp_file):
try:
os.remove(tmp_file)
except Exception:
pass

============================================================

HAUPTPROGRAMM

============================================================

def main():
print("Starte Akku2 RS485 -> TXT-Export")
print(f"RS485: {PORT} @ {BAUDRATE} Baud")
print(f"Ausgabe: {OUTPUT_FILE}")
print("-" * 70)

while True:
try:
with serial.Serial(PORT, BAUDRATE, timeout=0.1) as ser:
address, data = find_active_battery(ser)
if address is not None and data is not None:
max_capacity = update_capacity_history(address, data)
write_file(address, data, max_capacity)
print(
f"[{datetime.now().strftime('%H:%M:%S')}] Daten erfolgreich in"
f" {OUTPUT_FILE} geschrieben (Adresse {address})"
)
else:
print("[RS485] Kein aktiver Akku gefunden")
time.sleep(INTERVAL)
except Exception as e:
print(f"[FEHLER] {e}")
time.sleep(5)

if name == "main":
main()

Ausgabe liefert dann das:

timestamp=2026-09-12 22:50:09
address=2
voltage=49.023
current=-3.4
power=-166.7
cell_count=15
cell_min=3.255
cell_max=3.271
cell_delta=0.016
cell_delta_mv=16
cell1=3.269
cell2=3.269
cell3=3.269
cell4=3.255
cell5=3.270
cell6=3.268
cell7=3.269
cell8=3.270
cell9=3.269
cell10=3.268
cell11=3.268
cell12=3.271
cell13=3.270
cell14=3.270
cell15=3.268
temp_count=5
temp_min=25.9
temp_max=26.4
temp_avg=26.1
temp1=26.4
temp2=26.0
temp3=26.1
temp4=25.9
temp5=26.1
remaining_raw=1590
total_raw=4501
remaining_ah=15.90
total_ah=45.01
remaining_kwh=0.779
total_kwh=2.207
user_defined=2
cycles=26
soc=35.3
soc_source=bms_capacity_ratio
soh=unknown
soh_source=waiting_for_reference
remaining_hex=0636021195001A
raw_info=00020F0CC50CC50CC50CB70CC60CC40CC50CC60CC50CC40CC40CC70CC60CC60CC4050BB30BAF0BB00BAE0BB0FFDEBF7F0636021195001A