docs: Test-Abschnitt reparieren + sim_charge.py aufnehmen

- README: leerer Code-Block im Abschnitt "Test ohne Auto" ersetzt durch
  sim_charge.py-Beispiele und die mosquitto_pub-Sequenz.
- collector/sim_charge.py: Test-Skript zum Simulieren einer Ladung über MQTT.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-05 13:13:41 +02:00
parent 81ef0a82ce
commit ca708b5dda
2 changed files with 102 additions and 1 deletions
+10 -1
View File
@@ -49,7 +49,16 @@ Web-App danach unter `http://<server>:8082/` (Port via `WEB_PORT` in `.env`).
## Test ohne Auto
Eine Ladung lässt sich per `mosquitto_pub` simulieren (Reihenfolge beachten):
Am einfachsten über das mitgelieferte Skript `collector/sim_charge.py` (liest
Broker/Topic aus `.env`, sendet `car=2 → wh → cdi → car=4`):
```bash
.venv/bin/python collector/sim_charge.py # 5230 Wh, 3600 s
.venv/bin/python collector/sim_charge.py --wh 8000 --duration 5400
.venv/bin/python collector/sim_charge.py --end idle # mit car=1 beenden
```
Alternativ direkt per `mosquitto_pub` (Reihenfolge beachten):
```bash
mosquitto_pub -h nuccy -t go-eCharger/310014/car -m 2
+92
View File
@@ -0,0 +1,92 @@
#!/usr/bin/env python3
"""Simuliert eine go-e-Wallbox-Ladung über MQTT, um den Collector zu testen.
Sendet nacheinander:
car=2 (Start) -> wh -> cdi -> car=4 (Ende)
Broker/User/Topic werden wie beim Collector aus der .env gelesen.
Beispiele:
.venv/bin/python collector/sim_charge.py # 5230 Wh, 3600 s
.venv/bin/python collector/sim_charge.py --wh 8000 --duration 5400
.venv/bin/python collector/sim_charge.py --end idle # mit car=1 beenden
.venv/bin/python collector/sim_charge.py --no-finish # Session offen lassen
"""
import argparse
import json
import os
import sys
import time
import paho.mqtt.client as mqtt
from dotenv import load_dotenv
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 parse_args() -> argparse.Namespace:
p = argparse.ArgumentParser(description="Test-Ladung über MQTT simulieren")
p.add_argument("--wh", type=float, default=5230.5, help="geladene Energie in Wh")
p.add_argument("--duration", type=int, default=3600, help="Ladedauer in Sekunden")
p.add_argument("--end", choices=["complete", "idle"], default="complete",
help="Abschluss über car=4 (complete) oder car=1 (idle)")
p.add_argument("--no-finish", action="store_true",
help="Session nicht abschließen (kein car=1/4 senden)")
p.add_argument("--delay", type=float, default=0.7,
help="Pause zwischen den Nachrichten in Sekunden")
return p.parse_args()
def main() -> int:
args = parse_args()
client = mqtt.Client(
callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
client_id="auto-charger-sim",
)
if MQTT_USER:
client.username_pw_set(MQTT_USER, MQTT_PASS)
print(f"Verbinde zu {MQTT_HOST}:{MQTT_PORT} (Topic-Basis {TOPIC_BASE}) ...")
try:
client.connect(MQTT_HOST, MQTT_PORT, keepalive=60)
except OSError as exc:
print(f"FEHLER: Verbindung zum Broker fehlgeschlagen: {exc}")
return 1
client.loop_start()
end_car = 4 if args.end == "complete" else 1
steps: list[tuple[str, str]] = [
("car", "2"),
("wh", f"{args.wh}"),
("cdi", json.dumps({"type": 1, "value": args.duration * 1000})),
]
if not args.no_finish:
steps.append(("car", str(end_car)))
for key, payload in steps:
topic = f"{TOPIC_BASE}/{key}"
info = client.publish(topic, payload, qos=1)
info.wait_for_publish(timeout=5)
print(f" -> {topic} = {payload}")
time.sleep(args.delay)
client.loop_stop()
client.disconnect()
print("Fertig. Erwartete Collector-Ausgabe: 'Ladung gestartet ...' "
"und 'Ladung gespeichert: %.0f Wh, %d s'." % (args.wh, args.duration))
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except KeyboardInterrupt:
sys.exit(0)