ich dachte evt kann jemand den code gebrauchen. Ich lese damit einen Titan akku mit 2,4 kwh aus.
Code: Alles auswählen
#!/usr/bin/env python3
import serial
import time
import os
from datetime import datetime
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-direct.txt"
HISTORY_DIR = "/var/lib/akku2"
HISTORY_FILE = os.path.join(HISTORY_DIR, "capacity-history.txt")
------------------------------------------------------------
Bekannte Akku-Nennenergie
------------------------------------------------------------
BATTERY_NOMINAL_KWH = 2.4
def checksum(body):
total = sum(body.encode("ascii"))
return (-total) & 0xFFFF
def make_request(address):
Pylontech/PACE Get Analog Value
body = "20%02X4642E002%02X" % (address, address)
chk = checksum(body)
return ("~" + body + "%04X\r" % chk).encode("ascii")
def read_response(ser, address):
ser.reset_input_buffer()
request = make_request(address)
ser.write(request)
ser.flush()
deadline = time.time() + RESPONSE_TIMEOUT
data = bytearray()
while time.time() < deadline:
chunk = ser.read(256)
if chunk:
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"
)
except Exception:
return None
start = text.find("~")
if start < 0:
return None
text = text[start:]
end = text.find("\r")
if end < 0:
return None
frame = text[:end]
if len(frame) < 20:
return None
body = frame[1:]
received_checksum = body[-4:]
payload = body[:-4]
try:
calculated_checksum = checksum(payload)
if int(
received_checksum,
16
) != calculated_checksum:
return None
except Exception:
return None
try:
info = payload[12:]
except Exception:
return None
return info
def parse_response(info):
try:
if len(info) < 10:
return None
# --------------------------------------------------------
# INFO Header
# --------------------------------------------------------
info_flag = int(
info[0:2],
16
)
pack_id = int(
info[2:4],
16
)
cell_count = int(
info[4:6],
16
)
pos = 6
# --------------------------------------------------------
# Zellen
# --------------------------------------------------------
cells = []
for _ in range(cell_count):
if pos + 4 > len(info):
return None
raw = int(
info[pos:pos + 4],
16
)
cells.append(
raw / 1000.0
)
pos += 4
# --------------------------------------------------------
# Temperaturen
# --------------------------------------------------------
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
)
temp = (
raw - 2731
) / 10.0
temps.append(temp)
pos += 4
# --------------------------------------------------------
# Strom
# --------------------------------------------------------
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
# --------------------------------------------------------
# Spannung
# --------------------------------------------------------
if pos + 4 > len(info):
return None
voltage_raw = int(
info[pos:pos + 4],
16
)
voltage = (
voltage_raw / 1000.0
)
pos += 4
# --------------------------------------------------------
# Kapazitäts-/Zusatzdaten
#
# 2 Byte remaining
# 1 Byte user-defined
# 2 Byte total
# 2 Byte cycles
# --------------------------------------------------------
remaining_hex = info[pos:]
remaining_raw = None
total_raw = None
user_defined = None
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
# --------------------------------------------------------
# Wir interpretieren die BMS-Kapazitätswerte aktuell
# als 0.01 Ah.
#
# Beispiel:
# 4501 -> 45.01 Ah
# 1623 -> 16.23 Ah
# --------------------------------------------------------
remaining_ah = None
total_ah = None
if remaining_raw is not None:
remaining_ah = (
remaining_raw / 100.0
)
if total_raw is not None:
total_ah = (
total_raw / 100.0
)
# --------------------------------------------------------
# Energie
# --------------------------------------------------------
remaining_kwh = None
total_kwh = None
if remaining_ah is not None:
remaining_kwh = (
remaining_ah *
voltage /
1000.0
)
if total_ah is not None:
total_kwh = (
total_ah *
voltage /
1000.0
)
# --------------------------------------------------------
# SOC
#
# Direkt aus den beiden BMS-Kapazitätswerten.
# --------------------------------------------------------
soc = None
if (
remaining_raw is not None
and total_raw is not None
and total_raw > 0
and remaining_raw <= total_raw
):
soc = (
remaining_raw /
total_raw *
100.0
)
# Begrenzen
if soc is not None:
soc = max(
0.0,
min(
100.0,
soc
)
)
# --------------------------------------------------------
# Akku aktiv?
# --------------------------------------------------------
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_kwh,
"total_kwh": total_kwh,
"user_defined": user_defined,
"cycles": cycles,
"soc": soc,
"raw_info": info,
}
except Exception:
return None
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:
line = line.strip()
if not line:
continue
parts = line.split(";")
if len(parts) < 3:
continue
try:
value = float(
parts[2]
)
if (
maximum is None
or value > maximum
):
maximum = value
except Exception:
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
)
# --------------------------------------------------------
# Nur neue Werte protokollieren, wenn sie sich ändern.
# Das verhindert eine riesige Datei.
# --------------------------------------------------------
last_value = None
if os.path.exists(
HISTORY_FILE
):
try:
with open(
HISTORY_FILE,
"r"
) as f:
lines = f.readlines()
if lines:
last = lines[-1].strip()
parts = last.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(
"%s;%d;%.2f;%d\n"
% (
timestamp,
address,
total_ah,
cycles
)
)
# --------------------------------------------------------
# Bisheriges Maximum lesen
# --------------------------------------------------------
max_capacity = (
read_max_capacity()
)
# Aktuellen Wert berücksichtigen
if (
max_capacity is None
or total_ah > max_capacity
):
max_capacity = total_ah
return max_capacity
except Exception:
return None
def find_active_battery(ser):
for address in range(
FIRST_ADDRESS,
LAST_ADDRESS + 1
):
info = read_response(
ser,
address
)
if info is None:
time.sleep(
DELAY_BETWEEN_REQUESTS
)
continue
data = parse_response(
info
)
if (
data
and data["active"]
):
return address, data
time.sleep(
DELAY_BETWEEN_REQUESTS
)
return None, None
def write_file(
address,
data,
max_capacity
):
cells = data["cells"]
temps = data["temps"]
timestamp = datetime.now().strftime(
"%Y-%m-%d %H:%M:%S"
)
lines = []
# ------------------------------------------------------------
# Grunddaten
# ------------------------------------------------------------
lines.append(
"timestamp=" + timestamp
)
lines.append(
"address=%d" % address
)
lines.append(
"voltage=%.3f" %
data["voltage"]
)
lines.append(
"current=%.1f" %
data["current"]
)
lines.append(
"power=%.1f" %
data["power"]
)
# ------------------------------------------------------------
# Zellen
# ------------------------------------------------------------
lines.append(
"cell_count=%d" %
len(cells)
)
if cells:
cell_min = min(cells)
cell_max = max(cells)
cell_delta = (
cell_max -
cell_min
)
lines.append(
"cell_min=%.3f" %
cell_min
)
lines.append(
"cell_max=%.3f" %
cell_max
)
lines.append(
"cell_delta=%.3f" %
cell_delta
)
lines.append(
"cell_delta_mv=%d" %
round(
cell_delta * 1000
)
)
for i, value in enumerate(
cells,
1
):
lines.append(
"cell%d=%.3f" %
(i, value)
)
# ------------------------------------------------------------
# Temperaturen
# ------------------------------------------------------------
lines.append(
"temp_count=%d" %
len(temps)
)
if temps:
lines.append(
"temp_min=%.1f" %
min(temps)
)
lines.append(
"temp_max=%.1f" %
max(temps)
)
lines.append(
"temp_avg=%.1f" %
(
sum(temps) /
len(temps)
)
)
for i, value in enumerate(
temps,
1
):
lines.append(
"temp%d=%.1f" %
(i, value)
)
# ------------------------------------------------------------
# Kapazität
# ------------------------------------------------------------
if data["remaining_raw"] is not None:
lines.append(
"remaining_raw=%d" %
data["remaining_raw"]
)
if data["total_raw"] is not None:
lines.append(
"total_raw=%d" %
data["total_raw"]
)
if data["remaining_ah"] is not None:
lines.append(
"remaining_ah=%.2f" %
data["remaining_ah"]
)
if data["total_ah"] is not None:
lines.append(
"total_ah=%.2f" %
data["total_ah"]
)
if data["remaining_kwh"] is not None:
lines.append(
"remaining_kwh=%.3f" %
data["remaining_kwh"]
)
if data["total_kwh"] is not None:
lines.append(
"total_kwh=%.3f" %
data["total_kwh"]
)
if data["user_defined"] is not None:
lines.append(
"user_defined=%d" %
data["user_defined"]
)
if data["cycles"] is not None:
lines.append(
"cycles=%d" %
data["cycles"]
)
# ------------------------------------------------------------
# SOC
# ------------------------------------------------------------
if data["soc"] is not None:
lines.append(
"soc=%.1f" %
data["soc"]
)
lines.append(
"soc_source=bms_capacity_ratio"
)
else:
lines.append(
"soc=unknown"
)
lines.append(
"soc_source=unknown"
)
# ------------------------------------------------------------
# SOH
#
# Vergleich aktuelle BMS-Kapazität mit dem bisher höchsten
# beobachteten Wert.
# ------------------------------------------------------------
soh = None
if (
max_capacity is not None
and max_capacity > 0
and data["total_ah"] is not None
):
soh = (
data["total_ah"] /
max_capacity *
100.0
)
soh = max(
0.0,
min(
100.0,
soh
)
)
if soh is not None:
lines.append(
"soh=%.1f" % soh
)
lines.append(
"soh_source=observed_max_capacity"
)
else:
lines.append(
"soh=unknown"
)
lines.append(
"soh_source=waiting_for_reference"
)
if max_capacity is not None:
lines.append(
"max_observed_capacity_ah=%.2f"
% max_capacity
)
# ------------------------------------------------------------
# Rohdaten
# ------------------------------------------------------------
lines.append(
"remaining_hex=" +
data["remaining_hex"]
)
lines.append(
"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:
try:
if os.path.exists(
tmp_file
):
os.remove(
tmp_file
)
except Exception:
pass
def main():
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
)
time.sleep(
INTERVAL
)
except Exception:
time.sleep(5)
if name == "main":
main()Code: Alles auswählen
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