"""Testdaten fuer die Skoda-Seite. NUR zur Pruefung - raeumt selbst wieder auf. python3 skoda_testdaten.py fuellen python3 skoda_testdaten.py loeschen Erzeugt ein Jahr Fahrzeugleben: Fahrten zwischen ein paar Orten, Ladungen an der eigenen Wallbox mit mitlaufendem Zaehler, eine Schnellladung unterwegs, dazu stuendliche Ruhezeilen. Die Batterie altert absichtlich um zwei Prozent im Jahr - daran laesst sich pruefen, ob die Trendrechnung der Seite anschlaegt. """ import sys, math, random, datetime sys.path.insert(0, "/volume1/homes/wagner/SolarManager") import mysql.connector as mc import konfig random.seed(1337) ORTE = [ ("Zuhause", 47.577994, 10.262337), ("Marktoberdorf Zentrum", 47.779000, 10.616000), ("Kaufbeuren", 47.882000, 10.622000), ("Fuessen", 47.571000, 10.701000), ("Kempten", 47.727000, 10.313000), ] KAP0 = 79.0 # nutzbare kWh je 100 Prozentpunkte (Batterieseite), zu Beginn ALTERN = 0.02 # Anteil, den die Kapazitaet je Jahr verliert WIRKGRAD = 0.91 # was vom Wallbox-Strom im Akku ankommt def verbinden(): return mc.connect(**konfig.datenbank()) def fuellen(c): cur = c.cursor() cur.execute("SELECT COUNT(*) FROM skoda") n = cur.fetchone()[0] if n: print("Tabelle ist nicht leer (%d Zeilen) - abgebrochen." % n) return zeilen = [] jetzt = datetime.datetime.now().replace(minute=0, second=0, microsecond=0) tag = (jetzt - datetime.timedelta(days=365)).replace(hour=0) ende = jetzt - datetime.timedelta(hours=2) soc = 72 odo = 18400 wh = 1_240_000 # Wallbox-Zaehlerstand in Wh whog = 799_524 ort = 0 def zeile(zeitpunkt, **kw): z = dict(datetime=zeitpunkt, soc=soc, range_m=int(soc * 4.4), chg_state="CONNECT_CABLE", chg_type="OFF", chg_kw=0.0, chg_rem_min=0, target_soc=80, care_mode=1, chg_mode="MANUAL", max_ac="MAXIMUM", max_ac_a=16, locked=1, doors_locked="YES", doors="CLOSED", windows="CLOSED", lights="OFF", sunroof="CLOSED", trunk="CLOSED", bonnet="CLOSED", odo_km=odo, park_state="PARKED", lat=ORTE[ort][1], lon=ORTE[ort][2], address=ORTE[ort][0], ac_state="OFF", aux_state="OFF", vent_state="UNSUPPORTED", wb_kw=0.0, wb_plug=0, wb_wh_total=wh, wbog_kw=0.0, wbog_plug=0, wbog_wh_total=whog, pv_kw=0.0, grid_kw=0.4, http_status=200, rl_remaining=17) z.update(kw) zeilen.append(z) # Tagesweise, damit die Ereignisse zu den Uhrzeiten passen, zu denen sie # gehoeren. Ein festes Stundenraster trifft eine bestimmte Stunde je nach # Schrittweite naemlich nie. start = tag while tag < ende: jahre = (tag - start).days / 365.25 kap = KAP0 * (1 - ALTERN * jahre) zeile(tag + datetime.timedelta(hours=2)) zeile(tag + datetime.timedelta(hours=6)) # --- Fahrt an Werktagen am Vormittag ----------------------------- gefahren = False if tag.weekday() < 6 and soc > 25 and random.random() < 0.55: gefahren = True t = tag + datetime.timedelta(hours=9, minutes=random.randint(0, 50)) ziel = random.randrange(1, len(ORTE)) km = random.randint(18, 95) fahrzeit = max(20, int(km * 1.1)) zeile(t, park_state="IN_MOTION", locked=0, doors_locked="NO", lights="ON") dsoc = min(soc - 8, max(3, round(km * random.uniform(0.17, 0.24) / kap * 100))) odo += km soc -= dsoc ort = ziel t += datetime.timedelta(minutes=fahrzeit) zeile(t, lights="OFF") # Auswaerts gelegentlich schnellladen - der Hauszaehler steht dann # still, die Ladung faellt aus der Kapazitaetsrechnung heraus. if soc < 30 and random.random() < 0.6: for i in range(9): soc = min(80, soc + 5) zeile(t + datetime.timedelta(minutes=4 * i), chg_state="CHARGING", chg_type="DC", chg_kw=118.0, chg_rem_min=(80 - soc) * 2) t += datetime.timedelta(minutes=40) zeile(t) # Rueckfahrt am Nachmittag t += datetime.timedelta(hours=random.randint(1, 5)) zeile(t, park_state="IN_MOTION", locked=0, doors_locked="NO", lights="ON") dsoc = min(soc - 6, max(3, round(km * 0.2 / kap * 100))) odo += km soc -= dsoc ort = 0 t += datetime.timedelta(minutes=fahrzeit) zeile(t, lights="OFF", trunk="OPEN") # Kofferraum kurz auf zeile(t + datetime.timedelta(minutes=10)) # --- Laden zuhause, wenn es sich lohnt --------------------------- if ort == 0 and soc < 50: t = tag + datetime.timedelta(hours=(17 if gefahren else 10), minutes=random.randint(0, 40)) ziel_soc = 80 if random.random() < 0.8 else 100 takt, kw = 240, 10.9 # Der Zaehler misst vor dem Ladegeraet, das Fahrzeug meldet # dahinter. Der Unterschied ist der Ladewirkungsgrad - genau den # soll die Auswertung aus den Daten zurueckgewinnen. kw_auto = kw * WIRKGRAD schritt_wh = kw * takt / 3.6 # Wallbox-Seite schritt_akku = kw_auto * takt / 3.6 # was im Akku ankommt i = 0 while soc < ziel_soc and i < 400: pv = max(0.0, min(kw, random.gauss(6.5, 3.5))) zeile(t + datetime.timedelta(seconds=takt * i), chg_state="CHARGING", chg_type="AC", chg_kw=round(kw_auto, 2), chg_rem_min=int(max(0, ziel_soc - soc) * kap / 100 / kw_auto * 60), wb_kw=round(kw, 2), wb_plug=1, wb_wh_total=int(wh), pv_kw=round(pv, 2), grid_kw=round(kw - pv, 2)) wh += schritt_wh soc += 100 * (schritt_akku / 1000) / kap i += 1 soc = int(round(min(soc, 100))) t += datetime.timedelta(seconds=takt * i) zeile(t, chg_state="CONSERVING", chg_type="OFF", wb_plug=1, wb_wh_total=int(wh)) zeile(tag + datetime.timedelta(hours=22)) tag += datetime.timedelta(days=1) # Der letzte Stand soll ein bisschen was zeigen: Fenster auf, Licht aus. zeilen[-1]["windows"] = "OPEN" zeilen[-1]["locked"] = None zeilen[-1]["doors_locked"] = "UNKNOWN" spalten = list(zeilen[0].keys()) sql = ("INSERT INTO skoda (" + ",".join("`%s`" % s for s in spalten) + ") VALUES (" + ",".join(["%s"] * len(spalten)) + ")") cur.executemany(sql, [tuple(z[s] for s in spalten) for z in zeilen]) c.commit() print("%d Zeilen eingefuegt, %s bis %s" % (len(zeilen), zeilen[0]["datetime"], zeilen[-1]["datetime"])) print("Zaehler von 1.240.000 auf %d Wh (%.0f kWh geladen)" % (wh, (wh - 1_240_000) / 1000)) def loeschen(c): cur = c.cursor() cur.execute("DELETE FROM skoda") cur.execute("ALTER TABLE skoda AUTO_INCREMENT=1") cur.execute("DELETE FROM skoda_raw") cur.execute("ALTER TABLE skoda_raw AUTO_INCREMENT=1") c.commit() cur.execute("SELECT COUNT(*) FROM skoda") print("geloescht, Zeilen jetzt:", cur.fetchone()[0]) if __name__ == "__main__": c = verbinden() if len(sys.argv) > 1 and sys.argv[1] == "loeschen": loeschen(c) else: fuellen(c)