adminandClaude Opus 5
8c62455b2c
- gatherSkodaData.py liest bis zu drei API-Schluessel und nutzt sie reihum;
das Kontingent gilt gemessen fuers ganze Konto, daher feste Aufteilung
14 Abruf / 4 Befehle / 2 Reserve
- nach einem Befehl aus der Weboberflaeche wird einmal ausser der Reihe
nachgesehen
- Wallbox-Verlauf waehrend einer Ladung nach skoda_ladepunkte, dazu
skoda_ladepunkte_nachtragen.py fuer Ladungen aus EnergyFlow
- evPlug ist jetzt ein Wahrheitswert: bool("no car") war immer True
- skoda.conf.example beschreibt das gemeinsame Format
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
491 lines
20 KiB
Python
491 lines
20 KiB
Python
"""Pruefung der Zuordnung in gatherSkodaData.py ohne Fahrzeug und ohne Netz.
|
|
|
|
Das Auto kommt erst naechste Woche. Bis dahin laesst sich wenigstens
|
|
sicherstellen, dass eine Antwort richtig zerlegt wird und dass die Faelle
|
|
nicht abstuerzen, die im Betrieb sicher vorkommen: fehlende Teilbereiche,
|
|
UNKNOWN statt eines Werts, Fahrzeug in Bewegung, leere Antwort.
|
|
|
|
Dazu die Buchfuehrung ueber das Kontingent, die sich sonst nur im Betrieb
|
|
zeigen wuerde - und dort erst als 429: dass es eingehalten wird, dass die
|
|
Schluessel reihum drankommen und dass ein abgewiesener die anderen nicht
|
|
mitnimmt. Abschnitt 13 spielt dafuer einen ganzen Tag Dauerladen durch.
|
|
|
|
python3 skoda_test.py
|
|
"""
|
|
|
|
import datetime
|
|
import time
|
|
import gatherSkodaData as g
|
|
|
|
# Nie in die echte Datenbank: gatherData() stoesst beim Laden das Schreiben
|
|
# des Wallbox-Verlaufs an, und diese Pruefung laeuft auch auf der NAS.
|
|
geschrieben = []
|
|
g._ladepunkteSchreiben = lambda punkte: geschrieben.extend(punkte)
|
|
|
|
|
|
fehler = []
|
|
|
|
def pruefe(was, ist, soll):
|
|
if ist != soll:
|
|
fehler.append(was+": "+repr(ist)+" statt "+repr(soll))
|
|
|
|
def frisch():
|
|
g.ret = g.SkodaData()
|
|
return g.ret
|
|
|
|
_NAMEN = ("API_KEY", "API_KEY2", "API_KEY3")
|
|
|
|
def schluessel(anzahl, eigenerCmdKey=False):
|
|
"""So viele Schluessel setzen, als staenden sie in skoda.conf.
|
|
|
|
Ohne Datei und ohne Netz. Auch die gemeinsame Stundenbilanz wird geleert,
|
|
damit keine Probe die der vorigen erbt.
|
|
"""
|
|
g._schluessel[:] = []
|
|
werte = dict((_NAMEN[i], "schluessel"+str(i)) for i in range(anzahl))
|
|
if eigenerCmdKey:
|
|
werte["CMD_API_KEY"] = "steuerung"
|
|
g._schluesselUebernehmen(werte)
|
|
g._st["eigenerCmdKey"] = eigenerCmdKey
|
|
g._st["zuletzt"] = -1
|
|
g._st["sperre"] = 0.0
|
|
g._verbrauch.clear()
|
|
return g._schluessel
|
|
|
|
|
|
# --- 1. Vollstaendige Antwort ---------------------------------------------
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{
|
|
"vin":"TMBJB9NY5RF999999","name":"Elroy",
|
|
"charging":{"isVehicleInSavedLocation":True,
|
|
"carCapturedTimestamp":"2026-09-02T05:00:00Z",
|
|
"status":{"state":"CHARGING","chargeType":"AC","chargePowerInKw":10.8,
|
|
"battery":{"stateOfChargeInPercent":71,
|
|
"remainingCruisingRangeInMeters":249000}},
|
|
"settings":{"targetStateOfChargeInPercent":80,"chargingCareMode":"ACTIVATED"}}}})
|
|
pruefe("SoC", r.soc, 71)
|
|
pruefe("Reichweite km", r.range_km, 249.0)
|
|
pruefe("Rohwert bleibt roh", r.range_m, 249000)
|
|
pruefe("UTC nach Ortszeit", r.capChg, datetime.datetime(2026,9,2,7,0,0))
|
|
pruefe("Batterieschonung", r.careMode, True)
|
|
|
|
# --- 2. Reichweite, die schon in Kilometern kommt --------------------------
|
|
# Das Feld heisst Meter, das Beispiel der Spec sieht nach Kilometern aus.
|
|
# Beide Lesarten muessen zur selben Anzeige fuehren.
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"battery":
|
|
{"stateOfChargeInPercent":71,"remainingCruisingRangeInMeters":249}}}}})
|
|
pruefe("249 gilt als Kilometer", r.range_km, 249.0)
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"battery":
|
|
{"stateOfChargeInPercent":3,"remainingCruisingRangeInMeters":900}}}}})
|
|
pruefe("900 gilt als Kilometer", r.range_km, 900.0)
|
|
|
|
# --- 3. UNKNOWN ist kein Nein ---------------------------------------------
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"status":{
|
|
"overall":{"locked":"UNKNOWN","doorsLocked":"UNKNOWN","doors":"UNKNOWN",
|
|
"windows":"UNSUPPORTED","lights":"OFF"},
|
|
"detail":{"sunroof":"UNSUPPORTED","trunk":"CLOSED","bonnet":"CLOSED"}}}})
|
|
pruefe("UNKNOWN wird None", r.locked, None)
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"status":{
|
|
"overall":{"locked":"NO","doorsLocked":"NO","doors":"OPEN",
|
|
"windows":"CLOSED","lights":"OFF"},
|
|
"detail":{"sunroof":"CLOSED","trunk":"CLOSED","bonnet":"CLOSED"}}}})
|
|
pruefe("NO wird False", r.locked, False)
|
|
|
|
# --- 4. Fehlende Teilbereiche loeschen den letzten Stand nicht -------------
|
|
# Die API laesst einen Teil weg, wenn sie ihn gerade nicht bekommt, und legt
|
|
# stattdessen einen Eintrag in errors ab. Der alte Wert ist dann die bessere
|
|
# Auskunft als eine Null.
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"odometer":{"mileageInKm":12753},
|
|
"charging":{"status":{"battery":{"stateOfChargeInPercent":71}}}}})
|
|
g.uebernehmen({"vehicle":{"odometer":{"mileageInKm":12753}},
|
|
"errors":[{"type":"CHARGING_UNAVAILABLE"}]})
|
|
pruefe("SoC bleibt stehen", r.soc, 71)
|
|
pruefe("Fehler vermerkt", r.apiErrors, "CHARGING_UNAVAILABLE")
|
|
|
|
# --- 5. Fahrzeug in Bewegung: keine Koordinaten ---------------------------
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"parkingPosition":{"state":"PARKED",
|
|
"gpsCoordinates":{"latitude":47.5779944,"longitude":10.2623373}}}})
|
|
g.uebernehmen({"vehicle":{"parkingPosition":{"state":"IN_MOTION"}}})
|
|
pruefe("Zustand folgt", r.parkState, "IN_MOTION")
|
|
pruefe("letzte Position bleibt", r.lat, 47.577994)
|
|
|
|
# --- 6. Fahrenheit wird umgerechnet ---------------------------------------
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"airConditioning":{"state":"COOLING",
|
|
"targetTemperature":{"value":72.5,"unit":"FAHRENHEIT"}}}})
|
|
pruefe("Fahrenheit nach Celsius", r.acTargetC, 22.5)
|
|
|
|
# --- 7. Leere und kaputte Antworten stuerzen nicht ab ---------------------
|
|
r = frisch()
|
|
g.uebernehmen({})
|
|
g.uebernehmen({"vehicle":{}})
|
|
g.uebernehmen({"vehicle":{"charging":{},"status":{},"odometer":{},
|
|
"parkingPosition":{},"airConditioning":{}}})
|
|
pruefe("leer bleibt leer", r.soc, 0)
|
|
|
|
# --- 8. Ladeleistung faellt auf 0, wenn das Kabel raus ist ----------------
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"state":"CHARGING","chargePowerInKw":10.8}}}})
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"state":"CONNECT_CABLE"}}}})
|
|
pruefe("abgesteckt heisst 0 kW", r.chgKw, 0.0)
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"state":"CHARGING","chargePowerInKw":10.8}}}})
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"state":"CONSERVING"}}}})
|
|
pruefe("gesteckt behaelt den Wert", r.chgKw, 10.8)
|
|
|
|
# --- 9. Abrufintervall folgt der Lage ------------------------------------
|
|
r = frisch()
|
|
schluessel(1)
|
|
g._st["letzteAend"] = 0.0
|
|
g._haus["wbPlug"] = False; g._haus["wbKw"] = 0.0
|
|
r.chgState = "CHARGING"
|
|
pruefe("laedt", g._intervall(), float(g._I_LADEN))
|
|
r.chgState = ""
|
|
g._haus["wbPlug"] = True; g._haus["wbKw"] = 7.4
|
|
pruefe("Wallbox liefert", g._intervall(), float(g._I_LADEN))
|
|
g._haus["wbKw"] = 0.0
|
|
pruefe("gesteckt, laedt nicht", g._intervall(), g._I_GESTECKT)
|
|
g._haus["wbPlug"] = False
|
|
r.parkState = "IN_MOTION"
|
|
pruefe("unterwegs", g._intervall(), float(g._I_FAHRT))
|
|
r.parkState = "PARKED"
|
|
pruefe("lange nichts", g._intervall(), g._I_RUHE)
|
|
r.error = 3
|
|
pruefe("nach Fehlern zurueckhaltend", g._intervall(), 240.0)
|
|
r.error = 0
|
|
|
|
# Mehr Schluessel duerfen den Takt NICHT beschleunigen: das Kontingent gilt
|
|
# fuer alle zusammen (gemessen, siehe _LIMIT). Wer das hier wieder staffelt,
|
|
# baut 429-Antworten ein.
|
|
for lage, feld, wert in (("beim Laden", "chgState", "CHARGING"),
|
|
("unterwegs", "parkState", "IN_MOTION"),
|
|
("beim Parken", "parkState", "PARKED")):
|
|
r.chgState = ""
|
|
r.parkState = ""
|
|
setattr(r, feld, wert)
|
|
werte = []
|
|
for anzahl in (1, 2, 3):
|
|
schluessel(anzahl)
|
|
werte.append(g._intervall())
|
|
pruefe("Takt "+lage+" haengt nicht an der Zahl der Schluessel",
|
|
len(set(werte)), 1)
|
|
r.chgState = "CHARGING"
|
|
r.parkState = ""
|
|
schluessel(3)
|
|
pruefe("und beim Laden sind es 260 s", g._intervall(), 260.0)
|
|
|
|
# --- 10. Kontingent weitet den Abstand ------------------------------------
|
|
r = frisch()
|
|
sch = schluessel(1)[0]
|
|
pruefe("100 Anfragen in 60 s", round(g._kontingent(
|
|
{"RateLimit-Remaining":"100","RateLimit-Reset":"60"}, sch),3), 0.72)
|
|
pruefe("2 Anfragen in 600 s", round(g._kontingent(
|
|
{"RateLimit-Remaining":"2","RateLimit-Reset":"600"}, sch),1), 360.0)
|
|
pruefe("nichts mehr frei", g._kontingent(
|
|
{"RateLimit-Remaining":"0","RateLimit-Reset":"120"}, sch), 125.0)
|
|
pruefe("ohne Header kein Einfluss", g._kontingent({}, sch), 0.0)
|
|
|
|
# --- 11. Nur echte Aenderungen erzeugen eine Zeile ------------------------
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"battery":{"stateOfChargeInPercent":71}}}}})
|
|
a = g._signatur()
|
|
g._haus["pvKw"] = 8.2 # Hausseite aendert sich staendig
|
|
b = g._signatur()
|
|
pruefe("Hausseite loest nichts aus", a, b)
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"battery":{"stateOfChargeInPercent":72}}}}})
|
|
pruefe("SoC-Sprung loest aus", g._signatur() != a, True)
|
|
|
|
|
|
# --- 12. Plug-in-Hybrid: beide Antriebe werden getrennt gefuehrt ----------
|
|
# Welcher Skoda kommt, steht noch nicht fest. Ohne include-Parameter liefert
|
|
# die API von sich aus fuelStatus, wenn ein Verbrenner mitfaehrt - die
|
|
# Spalten dafuer muessen von Anfang an da sein, nachtragen laesst sich das
|
|
# nicht.
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"fuelStatus":{
|
|
"carType":"HYBRID","totalRangeInKm":620,
|
|
"carCapturedTimestamp":"2026-09-02T05:00:00Z",
|
|
"primaryEngineRange":{"engineType":"ELECTRIC","currentSoCInPercent":71,
|
|
"remainingRangeInKm":110},
|
|
"secondaryEngineRange":{"engineType":"GASOLINE","currentFuelLevelInPercent":87,
|
|
"remainingRangeInKm":510}}}})
|
|
pruefe("Fahrzeugart", r.carType, "HYBRID")
|
|
pruefe("Gesamtreichweite", r.totalRangeKm, 620.0)
|
|
pruefe("Antrieb 1 elektrisch", (r.eng1Type, r.eng1Soc), ("ELECTRIC", 71))
|
|
pruefe("Antrieb 2 Benzin", (r.eng2Type, r.eng2FuelPct), ("GASOLINE", 87))
|
|
pruefe("Tankanzeige fuer die Oberflaeche", r.fuelPct, 87)
|
|
pruefe("Zeitstempel Verbrenner", r.capFuel, datetime.datetime(2026,9,2,7,0,0))
|
|
|
|
# Umgekehrte Reihenfolge - die Spec legt nicht fest, welcher Motor wo steht.
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"fuelStatus":{"carType":"HYBRID",
|
|
"primaryEngineRange":{"engineType":"DIESEL","currentFuelLevelInPercent":42},
|
|
"secondaryEngineRange":{"engineType":"ELECTRIC","currentSoCInPercent":66}}}})
|
|
pruefe("Tank auch an erster Stelle", r.fuelPct, 42)
|
|
|
|
# Reiner Stromer: kein fuelStatus, alles bleibt leer.
|
|
r = frisch()
|
|
g.uebernehmen({"vehicle":{"charging":{"status":{"battery":
|
|
{"stateOfChargeInPercent":71}}}},"errors":[{"type":"FUEL_STATUS_UNSUPPORTED"}]})
|
|
pruefe("Stromer ohne Tank", (r.carType, r.fuelPct), ("", 0))
|
|
|
|
# --- 13. Das Kontingent wird unter Dauerlast eingehalten ------------------
|
|
# Die Doku nennt 20 Anfragen je Stunde *und Schluessel*. Beim Laden greift das
|
|
# kuerzeste Intervall - genau dort muss die Sperre halten. Simuliert wird ein
|
|
# ganzer Tag Dauerladen, ohne echte Zeit und ohne Netz.
|
|
|
|
def dauerlast(anzahl, eigenerCmdKey=False):
|
|
"""Einen Tag Dauerladen durchspielen. Liefert Zeitpunkte je Schluessel."""
|
|
schluessel(anzahl, eigenerCmdKey)
|
|
takt = float(max(g._I_MIN, g._I_LADEN))
|
|
punkte = dict((s.name, []) for s in g._schluessel)
|
|
t, gebremst = 0.0, 0
|
|
while t < 24*3600:
|
|
sch = g._waehlen(t) if g._budgetSperre(t) <= 0.0 else None
|
|
if sch is None:
|
|
# Nichts frei - genau das tut _abrufen() dann auch.
|
|
gebremst += 1
|
|
t += max(5.0, g._fruehesteFreigabe(t))
|
|
continue
|
|
g._gezaehlt(t)
|
|
punkte[sch.name].append(t)
|
|
t += takt
|
|
return punkte, gebremst
|
|
|
|
def schlimmstesFenster(zeiten):
|
|
"""Die meisten Anfragen, die in irgendeine Stunde fallen."""
|
|
return max([len([x for x in zeiten if t0 <= x < t0+3600])
|
|
for t0 in zeiten] or [0])
|
|
|
|
def alleZeiten(punkte):
|
|
zeiten = []
|
|
for z in punkte.values():
|
|
zeiten.extend(z)
|
|
return sorted(zeiten)
|
|
|
|
punkte, gebremst = dauerlast(1)
|
|
schlimmste = schlimmstesFenster(alleZeiten(punkte))
|
|
pruefe("nie mehr als das Limit je Stunde", schlimmste <= g._LIMIT, True)
|
|
pruefe("das eigene Budget wird eingehalten", schlimmste <= g._BUDGET, True)
|
|
pruefe("und auch ausgenutzt", schlimmste >= g._BUDGET - 1, True)
|
|
pruefe("der 260-s-Takt passt ohne Bremse", gebremst, 0)
|
|
|
|
# Der entscheidende Punkt: drei Schluessel duerfen NICHT mehr Anfragen
|
|
# ergeben. Sie teilen sich einen Zaehler - wer hier mehr misst, hat die
|
|
# Staffel wieder eingebaut und wird im Betrieb mit 429 bezahlt.
|
|
punkte3, gebremst3 = dauerlast(3)
|
|
pruefe("drei Schluessel halten dasselbe Limit",
|
|
schlimmstesFenster(alleZeiten(punkte3)) <= g._BUDGET, True)
|
|
pruefe("und bringen genauso viele Abrufe wie einer",
|
|
len(alleZeiten(punkte3)), len(alleZeiten(punkte)))
|
|
for name, zeiten in punkte3.items():
|
|
pruefe(name+" kommt reihum dran", len(zeiten) > 0, True)
|
|
|
|
# Ein eigener Schluessel fuer die Steuerung aendert am Kontingent nichts
|
|
# mehr - er trennt nur, wer womit fragt.
|
|
punkte, _ = dauerlast(1, eigenerCmdKey=True)
|
|
pruefe("eigener Befehlsschluessel gibt dem Abruf nicht mehr",
|
|
schlimmstesFenster(alleZeiten(punkte)) <= g._BUDGET, True)
|
|
|
|
# Reihum: jeder ist einmal dran, bevor einer ein zweites Mal drankommt. Nicht
|
|
# wegen des Kontingents - das ist gemeinsam -, sondern damit ein abgelaufener
|
|
# Schluessel auffaellt, statt jahrelang unbenutzt danebenzuliegen.
|
|
schluessel(3)
|
|
folge = []
|
|
for i in range(9):
|
|
sch = g._waehlen(i*300.0)
|
|
g._gezaehlt(i*300.0)
|
|
folge.append(sch.name)
|
|
pruefe("die Schluessel wechseln sich ab", folge, list(_NAMEN)*3)
|
|
|
|
# Ein abgewiesener Schluessel (401, 403) sperrt sich selbst - die uebrigen
|
|
# muessen weiterlaufen, sonst nimmt ein abgelaufener das ganze Fahrzeug mit.
|
|
schluessel(3)
|
|
g._schluessel[0].sperre = 1e9
|
|
pruefe("gesperrter wird uebersprungen",
|
|
[g._waehlen(float(i)).name for i in range(4)],
|
|
["API_KEY2", "API_KEY3"]*2)
|
|
for s in g._schluessel:
|
|
s.sperre = 500.0
|
|
pruefe("alle gesperrt: kein Abruf", g._waehlen(0.0), None)
|
|
pruefe("dafuer ein sauberer Wiedervorlagetermin",
|
|
g._fruehesteFreigabe(0.0), 500.0)
|
|
|
|
# Ein 429 dagegen gilt fuer alle - ein anderer Schluessel hilft da nicht.
|
|
schluessel(3)
|
|
g._st["sperre"] = 400.0
|
|
pruefe("429 haelt alle zurueck", g._budgetSperre(0.0), 400.0)
|
|
pruefe("und der Schluessel selbst bleibt unbescholten",
|
|
g._schluessel[0].sperre, 0.0)
|
|
|
|
# Wird ein Schluessel getauscht, behalten die anderen, was ueber sie bekannt
|
|
# ist - eine Sperre aus einem 401 und das Ablaufdatum. Sonst begaenne nach
|
|
# jedem Anfassen der Datei das Ausprobieren von vorn.
|
|
schluessel(3)
|
|
g._schluessel[0].sperre = 4711.0
|
|
g._schluessel[0].ablauf = datetime.datetime(2027, 5, 5)
|
|
g._schluesselUebernehmen({"API_KEY":"schluessel0", "API_KEY2":"schluessel1",
|
|
"API_KEY3":"frisch geholt"})
|
|
pruefe("Sperre ueberlebt den Tausch", g._schluessel[0].sperre, 4711.0)
|
|
pruefe("Ablaufdatum auch", g._schluessel[0].ablauf, datetime.datetime(2027, 5, 5))
|
|
pruefe("der neue faengt bei null an", g._schluessel[2].sperre, 0.0)
|
|
g._schluessel[:] = []
|
|
g._schluesselUebernehmen({"API_KEY":"x", "API_KEY2":"x"})
|
|
pruefe("derselbe Schluessel zweimal zaehlt einfach", len(g._schluessel), 1)
|
|
|
|
# --- 14. Header mit den echten Zahlen -------------------------------------
|
|
g.ret = g.SkodaData()
|
|
sch = schluessel(1)[0]
|
|
pruefe("19 Anfragen in 3400 s", round(g._kontingent(
|
|
{"RateLimit-Remaining":"19","RateLimit-Reset":"3400"}, sch),0), 215.0)
|
|
pruefe("Restkontingent gemerkt", sch.rest, 19)
|
|
|
|
# In die Historie geht die juengste Auskunft, nicht die Summe: die Schluessel
|
|
# teilen sich einen Zaehler und melden daher alle dieselbe Zahl.
|
|
drei = schluessel(3)
|
|
pruefe("ohne Auskunft bleibt es unbekannt", g._restGesamt(), -1)
|
|
drei[0].rest, drei[1].rest = 5, 7
|
|
g._st["zuletzt"] = 1
|
|
pruefe("der zuletzt benutzte Schluessel zaehlt", g._restGesamt(), 7)
|
|
drei[0].ablauf = datetime.datetime(2027, 1, 1)
|
|
drei[1].ablauf = datetime.datetime(2026, 10, 1)
|
|
pruefe("es zaehlt der Schluessel, der zuerst ablaeuft",
|
|
g._fruehesterAblauf(), datetime.datetime(2026, 10, 1))
|
|
|
|
# Abschnitt 15 soll wieder ohne Konfiguration laufen: keine Schluessel, keine
|
|
# Datei - sonst wuerde gatherData() auf der NAS die echte skoda.conf finden
|
|
# und einen Abruf anstossen, den diese Pruefung nicht will.
|
|
g._schluessel[:] = []
|
|
g._KONFIG = "/gibt/es/nicht/skoda.conf"
|
|
g._st["konfMtime"] = 0.0
|
|
|
|
# --- 15. Ladebeginn und Ladeende stossen einen Abruf an -------------------
|
|
# Bei 20 Anfragen je Stunde sind das die beiden wertvollsten Augenblicke:
|
|
# der Ladestand davor und danach traegt die Kapazitaetsrechnung.
|
|
import asyncio
|
|
g._st["ladenVorher"] = False
|
|
g._st["naechster"] = time.time() + 9999
|
|
asyncio.get_event_loop().run_until_complete(g.gatherData(wbKw=10.8, wbPlug=True))
|
|
pruefe("Beginn wird bemerkt", g._st["ladenVorher"], True)
|
|
pruefe("Abruf vorgezogen", g._st["naechster"] < time.time() + 120, True)
|
|
g._st["naechster"] = time.time() + 9999
|
|
asyncio.get_event_loop().run_until_complete(g.gatherData(wbKw=0.0, wbPlug=True))
|
|
pruefe("Ende wird bemerkt", g._st["ladenVorher"], False)
|
|
pruefe("Abruf wieder vorgezogen", g._st["naechster"] < time.time() + 120, True)
|
|
g._st["naechster"] = time.time() + 9999
|
|
asyncio.get_event_loop().run_until_complete(g.gatherData(wbKw=0.0, wbPlug=True))
|
|
pruefe("ohne Flanke kein Anstoss", g._st["naechster"] > time.time() + 1000, True)
|
|
|
|
# --- 16. Ein Befehl aus der Weboberflaeche zieht den Abruf vor ------------
|
|
# ajax/skodaCmd.php beruehrt nach einem angenommenen Befehl eine Datei, deren
|
|
# Aenderungszeit der Zeitpunkt des Befehls ist. Hier wird sie untergeschoben,
|
|
# statt den Webserver zu bemuehen.
|
|
import os
|
|
import tempfile
|
|
anstoss = os.path.join(tempfile.gettempdir(), "skoda_abruf_pruefung")
|
|
g._ANSTOSS_ORTE = (anstoss,)
|
|
if os.path.exists(anstoss):
|
|
os.remove(anstoss)
|
|
|
|
g._st["anstoss"] = None
|
|
g._st["naechster"] = time.time() + 9999
|
|
g._anstossPruefen()
|
|
pruefe("ohne Datei kein Anstoss", g._st["naechster"] > time.time()+1000, True)
|
|
|
|
# Beim ersten Blick nach dem Start wird nur gemerkt. Der letzte Befehl kann
|
|
# Tage her sein - dafuer einen Abruf auszugeben, waere Verschwendung.
|
|
open(anstoss, "w").close()
|
|
frueher = time.time() - 86400
|
|
os.utime(anstoss, (frueher, frueher))
|
|
g._anstossPruefen()
|
|
pruefe("alter Befehl wird nicht nachgeholt", g._st["naechster"] > time.time()+1000, True)
|
|
|
|
# Jetzt der Befehl.
|
|
befehl = time.time()
|
|
os.utime(anstoss, (befehl, befehl))
|
|
g._anstossPruefen()
|
|
pruefe("Abruf wird vorgezogen",
|
|
abs(g._st["naechster"] - (befehl + g._NACH_BEFEHL)) < 1.0, True)
|
|
pruefe("und zwar nicht sofort", g._st["naechster"] > befehl + 20, True)
|
|
|
|
# Ohne neuen Befehl bleibt es beim Plan - sonst haenge der Abruf fest.
|
|
g._st["naechster"] = befehl + 5000
|
|
g._anstossPruefen()
|
|
pruefe("ohne neuen Befehl kein zweiter Anstoss", g._st["naechster"], befehl + 5000)
|
|
|
|
# Ein frueher geplanter Abruf wird nicht nach hinten geschoben.
|
|
g._st["naechster"] = befehl + 1
|
|
spaeter = befehl + 10
|
|
os.utime(anstoss, (spaeter, spaeter))
|
|
g._anstossPruefen()
|
|
pruefe("nur vorziehen, nie verzoegern", g._st["naechster"], befehl + 1)
|
|
os.remove(anstoss)
|
|
|
|
# --- 17. Wallbox-Verlauf waehrend einer Ladung ----------------------------
|
|
# Punkte im Minutentakt, solange Strom fliesst, einer sofort am Anfang und
|
|
# einer mit 0 kW am Ende. Die Zeit wird von Hand gestellt.
|
|
for st in g._ladepunkt.values():
|
|
st["zuletzt"] = 0.0
|
|
st["aktiv"] = False
|
|
|
|
def haus(kw, wh, ogkw=0.0, ogwh=829520):
|
|
g._haus.update({"wbKw": kw, "wbWh": wh, "wbogKw": ogkw, "wbogWh": ogwh,
|
|
"pvKw": 8.0, "gridKw": -1.5})
|
|
|
|
t0 = 1757750400.0 # 13.09.2026, 10:00 Ortszeit ungefaehr
|
|
haus(0.0, 7751734)
|
|
pruefe("ruhende Wallbox schreibt nichts", g._ladepunkteSammeln(t0), [])
|
|
|
|
haus(4.94, 7751740)
|
|
p = g._ladepunkteSammeln(t0 + 3)
|
|
pruefe("Beginn sofort festgehalten", len(p), 1)
|
|
pruefe("mit Wallbox, Leistung und Zaehler", (p[0][1], p[0][2], p[0][3]),
|
|
("carport", 4.94, 7751740))
|
|
|
|
pruefe("drei Sekunden spaeter nichts", g._ladepunkteSammeln(t0 + 6), [])
|
|
pruefe("59 Sekunden nach dem Beginn noch nichts", g._ladepunkteSammeln(t0 + 62), [])
|
|
pruefe("nach 60 Sekunden der naechste Punkt", len(g._ladepunkteSammeln(t0 + 63)), 1)
|
|
|
|
haus(0.0, 7776854)
|
|
p = g._ladepunkteSammeln(t0 + 70)
|
|
pruefe("Ende mit 0 kW", [(x[1], x[2], x[3]) for x in p], [("carport", 0.0, 7776854)])
|
|
pruefe("danach wieder Ruhe", g._ladepunkteSammeln(t0 + 200), [])
|
|
|
|
# Ein Zaehlerstand 0 ist ein ausgefallener Abruf, kein Stand.
|
|
haus(11.0, 0)
|
|
p = g._ladepunkteSammeln(t0 + 300)
|
|
pruefe("Zaehler 0 wird NULL", p[0][3], None)
|
|
|
|
# Beide Wallboxen unabhaengig voneinander.
|
|
haus(0.0, 7776900, ogkw=7.2, ogwh=829600)
|
|
p = g._ladepunkteSammeln(t0 + 400)
|
|
pruefe("Carport endet, Obergeschoss beginnt",
|
|
sorted((x[1], x[2]) for x in p), [("carport", 0.0), ("og", 7.2)])
|
|
|
|
# Und gatherData() stoesst das Schreiben tatsaechlich an.
|
|
for st in g._ladepunkt.values():
|
|
st["zuletzt"] = 0.0
|
|
st["aktiv"] = False
|
|
del geschrieben[:]
|
|
g._st["naechster"] = time.time() + 9999
|
|
schleife = asyncio.get_event_loop()
|
|
schleife.run_until_complete(g.gatherData(wbKw=10.8, wbPlug=True, wbWh=7760000))
|
|
schleife.run_until_complete(asyncio.sleep(0.2)) # dem Hintergrund Zeit geben
|
|
pruefe("gatherData schreibt den Punkt", [(x[1], x[2]) for x in geschrieben],
|
|
[("carport", 10.8)])
|
|
|
|
|
|
print()
|
|
if fehler:
|
|
for f in fehler:
|
|
print("FEHLER "+f)
|
|
print("\n"+str(len(fehler))+" von "+str(len(fehler))+" Pruefungen fehlgeschlagen.")
|
|
else:
|
|
print("Alle Pruefungen bestanden.")
|