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
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()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