admin
81ef0a82ce
- web/lib/db.ts: pro Request frische readonly-Verbindung statt langlebigem Singleton. Eine gecachte WAL-Verbindung bekommt Writes des Collectors über den Docker-FUSE-Mount nicht mit -> Seite blieb leer. - collector.py: connect_async + reconnect_delay_set statt connect(), damit ein nicht erreichbarer Broker keinen Container-Crash-Loop auslöst, sondern in-process mit Backoff neu versucht wird. - docker-compose.yml: DB_PATH=/data/charges.db für den collector-Service erzwungen, damit ein relatives DB_PATH aus .env (lokaler Lauf) im Container nicht neben das gemountete Volume schreibt. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
139 lines
4.1 KiB
Python
139 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
"""MQTT-Collector: liest go-e Wallbox-Daten und speichert abgeschlossene
|
|
Ladungen als je eine Zeile in SQLite.
|
|
|
|
Session-Erkennung über den 'car'-State:
|
|
1=Idle, 2=Charging, 3=WaitCar, 4=Complete.
|
|
Eine Session startet beim Wechsel nach 2 (Charging) und wird beim Wechsel
|
|
nach 1 (Idle) oder 4 (Complete) abgeschlossen. Dauer kommt aus 'cdi', die
|
|
geladene Energie aus 'wh'.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
|
|
import paho.mqtt.client as mqtt
|
|
from dotenv import load_dotenv
|
|
|
|
import db
|
|
|
|
load_dotenv()
|
|
|
|
MQTT_HOST = os.environ.get("MQTT_HOST", "localhost")
|
|
MQTT_PORT = int(os.environ.get("MQTT_PORT", "1883"))
|
|
MQTT_USER = os.environ.get("MQTT_USER") or None
|
|
MQTT_PASS = os.environ.get("MQTT_PASS") or None
|
|
TOPIC_BASE = os.environ.get("MQTT_TOPIC_BASE", "go-eCharger/310014").rstrip("/")
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
print(f"[{datetime.now().isoformat(timespec='seconds')}] {msg}", flush=True)
|
|
|
|
|
|
class Session:
|
|
"""In-Memory-Zustand der aktuell laufenden Ladung."""
|
|
|
|
def __init__(self) -> None:
|
|
self.active = False
|
|
self.start_time = None
|
|
self.energy_wh = 0.0 # letzter wh-Wert
|
|
self.duration_ms = 0 # letzter cdi.value
|
|
|
|
def start(self) -> None:
|
|
self.active = True
|
|
self.start_time = datetime.now().isoformat(timespec="seconds")
|
|
self.energy_wh = 0.0
|
|
self.duration_ms = 0
|
|
log(f"Ladung gestartet um {self.start_time}")
|
|
|
|
def finish(self) -> None:
|
|
if not self.active:
|
|
return
|
|
end_time = datetime.now().isoformat(timespec="seconds")
|
|
duration_s = int(self.duration_ms / 1000)
|
|
if self.energy_wh > 0:
|
|
db.insert_charge(self.start_time, end_time, duration_s, self.energy_wh)
|
|
log(f"Ladung gespeichert: {self.energy_wh:.0f} Wh, {duration_s} s")
|
|
else:
|
|
log("Ladung ohne Energie (>0 Wh) verworfen")
|
|
self.active = False
|
|
|
|
|
|
session = Session()
|
|
|
|
|
|
def on_connect(client, userdata, flags, reason_code, properties=None):
|
|
log(f"Verbunden mit MQTT (rc={reason_code})")
|
|
for key in ("car", "cdi", "wh"):
|
|
topic = f"{TOPIC_BASE}/{key}"
|
|
client.subscribe(topic)
|
|
log(f"abonniert: {topic}")
|
|
|
|
|
|
def on_disconnect(client, userdata, flags, reason_code=None, properties=None):
|
|
log(f"Verbindung getrennt (rc={reason_code})")
|
|
|
|
|
|
def on_message(client, userdata, msg):
|
|
payload = msg.payload.decode("utf-8", errors="replace").strip()
|
|
key = msg.topic.rsplit("/", 1)[-1]
|
|
|
|
if key == "wh":
|
|
try:
|
|
session.energy_wh = float(payload)
|
|
except ValueError:
|
|
pass
|
|
|
|
elif key == "cdi":
|
|
# {"type":0|1,"value":<ms>} oder null
|
|
try:
|
|
data = json.loads(payload)
|
|
except (ValueError, TypeError):
|
|
data = None
|
|
if isinstance(data, dict) and data.get("type") == 1:
|
|
session.duration_ms = int(data.get("value") or 0)
|
|
|
|
elif key == "car":
|
|
try:
|
|
car = int(payload)
|
|
except ValueError:
|
|
return
|
|
if car == 2 and not session.active:
|
|
session.start()
|
|
elif car in (1, 4) and session.active:
|
|
session.finish()
|
|
# car == 3 (WaitCar/Pause): Session bleibt aktiv
|
|
|
|
|
|
def main() -> None:
|
|
db.init_db()
|
|
log(f"DB bereit: {db.get_db_path()}")
|
|
|
|
client = mqtt.Client(
|
|
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
|
|
client_id="auto-charger-collector",
|
|
)
|
|
if MQTT_USER:
|
|
client.username_pw_set(MQTT_USER, MQTT_PASS)
|
|
client.on_connect = on_connect
|
|
client.on_disconnect = on_disconnect
|
|
client.on_message = on_message
|
|
|
|
# connect_async + loop_forever: bei nicht erreichbarem Broker wird
|
|
# in-process mit Backoff neu versucht, statt den Prozess zu beenden
|
|
# (kein Container-Crash-Loop). Verbindet sich selbst, sobald der
|
|
# Broker erreichbar ist.
|
|
log(f"Verbinde zu {MQTT_HOST}:{MQTT_PORT} ...")
|
|
client.reconnect_delay_set(min_delay=1, max_delay=60)
|
|
client.connect_async(MQTT_HOST, MQTT_PORT, keepalive=60)
|
|
client.loop_forever(retry_first_connection=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except KeyboardInterrupt:
|
|
sys.exit(0)
|