diff --git a/backend/scripts/seed_jahr.py b/backend/scripts/seed_jahr.py new file mode 100644 index 0000000..2774d51 --- /dev/null +++ b/backend/scripts/seed_jahr.py @@ -0,0 +1,289 @@ +"""Simuliert ein Jahr Fuhrpark-Nutzung inkl. Wartungsplänen/-intervallen +(km- und zeitbasiert) - synthetische Testdaten für GUI-Review, NUR für die +separate Test-Instanz (eigene DB, kein Produktionsbezug). + +Ergänzt seed_monat.py um einen echten Fahrzeug-Fuhrpark mit wachsendem +Kilometerstand und automatisch ausgelösten Wartungsaufträgen (Wartungsplan/ +Wartungsintervall, nicht nur Ad-hoc-Wartungsaufträge wie im 30-Tage-Seed). + +Aufruf (auf dem Server, im venv der Test-Instanz): + cd /opt/mabea-test/backend && .venv/bin/python -m scripts.seed_jahr +""" + +import asyncio +import random +from datetime import date, datetime, timedelta, timezone +from decimal import Decimal + +import app.models # noqa: F401 (alle ORM-Modelle registrieren, sonst NoReferencedTableError) +from app.core.security import hash_password +from app.db.session import SessionLocal +from app.models.auth import Benutzer, BenutzerRolle, KnotenTyp, RolleTyp, Systemknoten +from app.models.fahrzeugdetails import Fahrzeugdetails +from app.models.lager import Bestand, Lagerort, LagerortTyp, Lagerplatz, Materialbewegung +from app.models.mangel import Mangel, MangelPrioritaet, MangelStatus +from app.models.objekt import Objekt +from app.models.kontrolle import Kontrolle, Kontrollposition, KontrollStatus +from app.models.stammdaten import Bereich, Material, MaterialTyp, Objekttyp, Standort, ZaehlerTyp +from app.models.verleih import ExterneStelle, ExterneStelleKategorie, Verleih +from app.models.wartung import ( + IntervallTyp, + Wartungsauftrag, + WartungsauftragStatus, + Wartungsintervall, + Wartungsplan, + WartungsplanPosition, +) + +random.seed(7) + +TAGE = 365 +START = datetime.now(timezone.utc) - timedelta(days=TAGE) + +KM_INTERVALL = 15000 +ZEIT_INTERVALL_MONATE = 12 +KM_PRO_TAG_MIN, KM_PRO_TAG_MAX = 40, 160 + + +def zeitpunkt(tag_offset: int, stunde: int = 9) -> datetime: + return START + timedelta(days=tag_offset, hours=stunde, minutes=random.randint(0, 59)) + + +async def main() -> None: + async with SessionLocal() as db: + knoten_row = (await db.execute(Systemknoten.__table__.select())).first() + if knoten_row is None: + knoten_obj = Systemknoten(name="Hauptserver", typ=KnotenTyp.haupt) + db.add(knoten_obj) + await db.flush() + knoten_id = knoten_obj.id + else: + knoten_id = knoten_row[0] + + standorte = [Standort(name=n) for n in ["Wache Nord", "Wache Süd", "Wache Ost"]] + db.add_all(standorte) + await db.flush() + + bereich_fzg = Bereich(name="Fuhrpark") + db.add(bereich_fzg) + await db.flush() + + typ_rtw = Objekttyp( + bereich_id=bereich_fzg.id, name="RTW", ist_zugfahrzeug=True, + zaehler_typ=ZaehlerTyp.kilometer, wartung_aktiv=True, + ressourcenbuchung_aktiv=True, + ) + typ_ktw = Objekttyp( + bereich_id=bereich_fzg.id, name="KTW", ist_zugfahrzeug=True, + zaehler_typ=ZaehlerTyp.kilometer, wartung_aktiv=True, + ressourcenbuchung_aktiv=True, + ) + db.add_all([typ_rtw, typ_ktw]) + await db.flush() + + material_liste = [ + Material(name="Verbandpäckchen", einheit="Stück", materialtyp=MaterialTyp.standard), + Material(name="Infusionslösung 500ml", einheit="Stück", materialtyp=MaterialTyp.ablauf_charge), + ] + db.add_all(material_liste) + await db.flush() + + benutzer_specs = [ + ("fschulz", "Frank Schulz", RolleTyp.mitarbeiter), + ("jhoffmann", "Julia Hoffmann", RolleTyp.mitarbeiter), + ("mbauer", "Martin Bauer", RolleTyp.materialverantwortlicher), + ("lkoenig", "Laura König", RolleTyp.leitungsverantwortlicher), + ("admin2", "Admin Fuhrpark", RolleTyp.administration), + ] + benutzer = [] + for login, name, rolle in benutzer_specs: + b = Benutzer(name=name, login=login, passwort_hash=hash_password("Test1234!"), aktiv=True) + db.add(b) + await db.flush() + db.add(BenutzerRolle(benutzer_id=b.id, rolle=rolle)) + benutzer.append(b) + await db.flush() + mitarbeiter = [b for b, (_, _, r) in zip(benutzer, benutzer_specs) if r == RolleTyp.mitarbeiter] + verantwortlicher = benutzer[2] + + werkstatt = ExterneStelle(name="Fuhrpark-Werkstatt Nord", kategorie=ExterneStelleKategorie.werkstatt) + db.add(werkstatt) + await db.flush() + + # --- Wartungspläne mit km- und zeitbasierten Intervallen ----------------- + plan_rtw = Wartungsplan(objekttyp_id=typ_rtw.id, name="RTW-Wartungsplan") + plan_ktw = Wartungsplan(objekttyp_id=typ_ktw.id, name="KTW-Wartungsplan") + db.add_all([plan_rtw, plan_ktw]) + await db.flush() + + for plan in (plan_rtw, plan_ktw): + inspektion = WartungsplanPosition( + wartungsplan_id=plan.id, wartungsart="Inspektion", + beschreibung="Herstellerinspektion nach Kilometerleistung", + ) + hu_uvv = WartungsplanPosition( + wartungsplan_id=plan.id, wartungsart="HU/UVV-Prüfung", + beschreibung="Jährliche Hauptuntersuchung + UVV-Prüfung", + ) + db.add_all([inspektion, hu_uvv]) + await db.flush() + db.add(Wartungsintervall( + wartungsplan_position_id=inspektion.id, typ=IntervallTyp.km, wert=KM_INTERVALL, + )) + db.add(Wartungsintervall( + wartungsplan_position_id=hu_uvv.id, typ=IntervallTyp.zeit, wert=ZEIT_INTERVALL_MONATE, + )) + await db.flush() + + # --- Fuhrpark: 3 RTW + 2 KTW mit individuellem Jahres-Kilometerprofil ---- + fahrzeuge = [] + for i in range(1, 4): + o = Objekt( + code=f"RTW-{i:02d}", name=f"RTW {i}", objekttyp_id=typ_rtw.id, + standort_id=random.choice(standorte).id, zustaendiger_server_id=knoten_id, + ) + db.add(o) + fahrzeuge.append((o, random.randint(15000, 40000))) + for i in range(1, 3): + o = Objekt( + code=f"KTW-{i:02d}", name=f"KTW {i}", objekttyp_id=typ_ktw.id, + standort_id=random.choice(standorte).id, zustaendiger_server_id=knoten_id, + ) + db.add(o) + fahrzeuge.append((o, random.randint(10000, 25000))) + await db.flush() + + fahrzeugdetails = {} + for objekt, start_km in fahrzeuge: + fd = Fahrzeugdetails( + objekt_id=objekt.id, + kennzeichen=f"XX-RD {objekt.id}23", + funkrufname=objekt.code, + baujahr=random.randint(2018, 2024), + kilometerstand=start_km, + betriebsstunden=None, + naechste_hu=(START + timedelta(days=ZEIT_INTERVALL_MONATE * 30)).date(), + naechste_uvv=(START + timedelta(days=ZEIT_INTERVALL_MONATE * 30)).date(), + ) + db.add(fd) + fahrzeugdetails[objekt.id] = {"obj": objekt, "km": start_km, "letzte_insp_km": start_km, "start_km": start_km} + await db.flush() + + lager_standort = Lagerort(name="Fuhrpark-Lager", typ=LagerortTyp.standort, standort_id=standorte[0].id) + db.add(lager_standort) + await db.flush() + lagerplatz = Lagerplatz(lagerort_id=lager_standort.id, code="FP-01") + db.add(lagerplatz) + await db.flush() + for m in material_liste: + db.add(Bestand(lagerplatz_id=lagerplatz.id, material_id=m.id, menge=Decimal(200))) + await db.flush() + + letzte_hu_uvv_am = {objekt.id: START for objekt, _ in fahrzeuge} + + for tag_offset in range(TAGE): + zp = zeitpunkt(tag_offset) + aktueller_tag = zp.date() + + # Kilometerstand je Fahrzeug fortschreiben (6/7 Tage im Einsatz). + if tag_offset % 7 != 0: + for objekt, _ in fahrzeuge: + info = fahrzeugdetails[objekt.id] + info["km"] += random.randint(KM_PRO_TAG_MIN, KM_PRO_TAG_MAX) + + # km-Inspektion auslösen, sobald Intervall seit letzter Inspektion voll ist. + for objekt, _ in fahrzeuge: + info = fahrzeugdetails[objekt.id] + if info["km"] - info["letzte_insp_km"] >= KM_INTERVALL: + auftrag = Wartungsauftrag( + objekt_id=objekt.id, wartungsart="Inspektion", + status=WartungsauftragStatus.erledigt, werkstatt_intern=False, + dienstleister_id=werkstatt.id, erstellt_am=zp, + erledigt_am=zp + timedelta(days=1), durchgefuehrt_von=verantwortlicher.id, + kosten=Decimal(random.randint(380, 650)), + kilometerstand_bei_erledigung=info["km"], + ) + db.add(auftrag) + info["letzte_insp_km"] = info["km"] + + # Jährliche HU/UVV, wenn 12 Monate seit letzter Prüfung vergangen sind. + for objekt, _ in fahrzeuge: + letzte = letzte_hu_uvv_am[objekt.id] + if (zp - letzte).days >= ZEIT_INTERVALL_MONATE * 30: + info = fahrzeugdetails[objekt.id] + auftrag = Wartungsauftrag( + objekt_id=objekt.id, wartungsart="HU/UVV-Prüfung", + status=WartungsauftragStatus.erledigt, werkstatt_intern=False, + dienstleister_id=werkstatt.id, erstellt_am=zp, + erledigt_am=zp + timedelta(days=2), durchgefuehrt_von=verantwortlicher.id, + kosten=Decimal(random.randint(150, 280)), + kilometerstand_bei_erledigung=info["km"], + ) + db.add(auftrag) + letzte_hu_uvv_am[objekt.id] = zp + + # Noch offener Wartungsauftrag für ein zufälliges Fahrzeug alle ~18 Tage + # (kleine Reparatur, nicht planbasiert). + if tag_offset % 18 == 0: + objekt, _ = random.choice(fahrzeuge) + db.add(Wartungsauftrag( + objekt_id=objekt.id, wartungsart="Reifenwechsel", + status=WartungsauftragStatus.offen, werkstatt_intern=True, + erstellt_am=zp, faelligkeit_am=aktueller_tag + timedelta(days=5), + )) + + # Kontrollen: wöchentlich ein Fahrzeug. + if tag_offset % 7 == 0: + objekt, _ = random.choice(fahrzeuge) + pruefer = random.choice(mitarbeiter) + k = Kontrolle( + erzeugt_von_server_id=knoten_id, objekt_id=objekt.id, benutzer_id=pruefer.id, + status=KontrollStatus.abgeschlossen, gestartet_am=zp, + beendet_am=zp + timedelta(minutes=20), + ) + db.add(k) + await db.flush() + for m in material_liste: + soll = Decimal(20) + ist = soll if random.random() > 0.1 else soll - Decimal(random.randint(1, 4)) + db.add(Kontrollposition( + kontrolle_id=k.id, material_id=m.id, + sollmenge_snapshot=soll, istmenge_erfasst=ist, abweichung=ist != soll, + )) + + # Mängel: alle ~20 Tage. + if tag_offset % 20 == 0: + objekt, _ = random.choice(fahrzeuge) + melder = random.choice(mitarbeiter) + status = random.choice([MangelStatus.neu, MangelStatus.in_bearbeitung, MangelStatus.erledigt]) + db.add(Mangel( + objekt_id=objekt.id, beschreibung=f"Auffälligkeit bei {objekt.name} gemeldet", + prioritaet=random.choice(list(MangelPrioritaet)), status=status, + gemeldet_von=melder.id, gemeldet_am=zp, + erledigt_von=verantwortlicher.id if status == MangelStatus.erledigt else None, + erledigt_am=zp + timedelta(days=3) if status == MangelStatus.erledigt else None, + )) + + # Lagerbewegung: alle 4 Tage Entnahme für Fahrzeugausstattung. + if tag_offset % 4 == 0: + m = random.choice(material_liste) + db.add(Materialbewegung( + material_id=m.id, menge=Decimal(random.randint(1, 6)), + von_lagerplatz_id=lagerplatz.id, benutzer_id=random.choice(mitarbeiter).id, + zeitpunkt=zp, grund="Nachfüllen Fahrzeugausstattung", + )) + + # Fahrzeugdetails auf den finalen Kilometerstand setzen (für Akte/Dashboard). + for objekt, _ in fahrzeuge: + info = fahrzeugdetails[objekt.id] + fd = await db.get(Fahrzeugdetails, objekt.id) + fd.kilometerstand = info["km"] + + await db.commit() + gesamt_km = sum(info["km"] - info["start_km"] for info in fahrzeugdetails.values()) + print(f"Seed fertig: {len(fahrzeuge)} Fahrzeuge, 365 Tage, insgesamt {gesamt_km} simulierte km.") + print("Login z.B.: admin2 / Test1234!") + + +if __name__ == "__main__": + asyncio.run(main())