admin
ca708b5dda
- 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>
93 lines
3.1 KiB
Python
93 lines
3.1 KiB
Python
#!/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)
|