Skoda: mehrere Schluessel, Wallbox-Ladeverlauf, Stecker als 1/0
- 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>
This commit is contained in:
Claude Opus 5
+275
-43
@@ -5,6 +5,11 @@ 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
|
||||
"""
|
||||
|
||||
@@ -12,6 +17,11 @@ 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 = []
|
||||
|
||||
@@ -23,6 +33,25 @@ 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()
|
||||
@@ -112,33 +141,55 @@ 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(), g._I_LADEN)
|
||||
pruefe("laedt", g._intervall(), float(g._I_LADEN))
|
||||
r.chgState = ""
|
||||
g._haus["wbPlug"] = True; g._haus["wbKw"] = 7.4
|
||||
pruefe("Wallbox liefert", g._intervall(), g._I_LADEN)
|
||||
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(), g._I_FAHRT)
|
||||
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"}),3), 0.72)
|
||||
{"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"}),1), 360.0)
|
||||
{"RateLimit-Remaining":"2","RateLimit-Reset":"600"}, sch),1), 360.0)
|
||||
pruefe("nichts mehr frei", g._kontingent(
|
||||
{"RateLimit-Remaining":"0","RateLimit-Reset":"120"}), 125.0)
|
||||
pruefe("ohne Header kein Einfluss", g._kontingent({}), 0.0)
|
||||
{"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()
|
||||
@@ -185,51 +236,134 @@ g.uebernehmen({"vehicle":{"charging":{"status":{"battery":
|
||||
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
|
||||
# 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.
|
||||
g._verbrauch.clear()
|
||||
zeitpunkte = []
|
||||
t = 0.0
|
||||
while t < 24*3600:
|
||||
zeitpunkte.append(t)
|
||||
g._gezaehlt(t)
|
||||
t += max(g._I_MIN, g._I_LADEN, g._budgetSperre(t))
|
||||
schlimmste = 0
|
||||
for i, t0 in enumerate(zeitpunkte):
|
||||
in_fenster = sum(1 for t1 in zeitpunkte[i:] if t1 - t0 < 3600)
|
||||
schlimmste = max(schlimmste, in_fenster)
|
||||
|
||||
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_GETEILT, True)
|
||||
pruefe("und auch ausgenutzt", schlimmste >= g._BUDGET_GETEILT - 3, 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)
|
||||
|
||||
# Mit eigenem Schluessel fuer die Steuerung darf mehr genutzt werden.
|
||||
g._st["eigenerCmdKey"] = True
|
||||
g._verbrauch.clear()
|
||||
for i in range(g._BUDGET_GETEILT):
|
||||
g._gezaehlt(2000.0 + i)
|
||||
pruefe("geteiltes Budget reicht dem Modul allein nicht",
|
||||
g._budgetSperre(2000.0 + g._BUDGET_GETEILT), 0.0)
|
||||
g._verbrauch.clear()
|
||||
for i in range(g._BUDGET_ALLEIN):
|
||||
g._gezaehlt(2000.0 + i)
|
||||
pruefe("eigenes Budget greift bei 18", g._budgetSperre(2000.0 + g._BUDGET_ALLEIN) > 3500, True)
|
||||
g._st["eigenerCmdKey"] = False
|
||||
g._verbrauch.clear()
|
||||
# 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 einzelner Anstoss ausser der Reihe darf das Budget nicht sprengen.
|
||||
g._verbrauch.clear()
|
||||
for i in range(g._BUDGET_GETEILT):
|
||||
g._gezaehlt(1000.0 + i)
|
||||
pruefe("Budget voll: Sperre greift", g._budgetSperre(1000.0+g._BUDGET_GETEILT) > 3500, True)
|
||||
g._verbrauch.clear()
|
||||
pruefe("Budget leer: keine Sperre", g._budgetSperre(0.0), 0.0)
|
||||
# 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"}),0), 215.0)
|
||||
pruefe("Restkontingent gemerkt", g.ret.rlRemaining, 19)
|
||||
{"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:
|
||||
@@ -248,6 +382,104 @@ 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:
|
||||
|
||||
Reference in New Issue
Block a user