From f4791cf7bb6df15ddd40f7f0e2206e277d834759 Mon Sep 17 00:00:00 2001 From: Jan Hoheisel Date: Thu, 25 Dec 2025 21:43:29 +0100 Subject: [PATCH] refactoring: Neue Architektur --- README.md | 206 +++++++++++------ STRUKTUR.md | 107 --------- ausgabe.py | 251 +++++++++++++++++++++ datenmodell.py | 114 ++++++++++ elterndienstplaner.py | 499 +++++++++--------------------------------- typen.py | 29 +++ 6 files changed, 634 insertions(+), 572 deletions(-) create mode 100644 ausgabe.py create mode 100644 datenmodell.py create mode 100644 typen.py diff --git a/README.md b/README.md index ba28ccf..4b8fac4 100644 --- a/README.md +++ b/README.md @@ -1,95 +1,163 @@ -Elterndienstplaner +# Elterndienstplaner -syntax: ./elterndienstplaner.py { } +Automatische Zuteilung von Elterndiensten im Kinderladen unter Berücksichtigung von Fairness und Präferenzen. -Der Elterndienstplaner hilft bei der Zuteilung von Elterndiensten zu Eltern. -Zur Identifizierung der Eltern dient der Name des Kindes/der Kinder. +## Verwendung -Es gibt diese Dienste: - - F: Frühstücksdienst (täglich) - - P: Putznotdienst (täglich) - - E: Essensausgabenotdienst (täglich) - - K: Kochen (alle 2 wochen, Datum manuell festgelegt) - - A: Elternabend (2 Eltern, Datum manuell festgelegt) +```bash +./elterndienstplaner.py [] +``` -Die Planung erfolgt immer für einen Kalendermonat. +**Parameter:** +- `eingabe.csv`: Benötigte Dienste und Eltern-Präferenzen für den Planungsmonat +- `eltern.csv`: Dienstfaktoren der Eltern (Anzahl betreuter Kinder) +- `ausgabe.csv`: Hier wird die Zuteilung geschrieben +- `vorherige-ausgaben.csv` (optional): Historische Daten für Fairness über das Jahr -Die Eltern geben an, an welchen Tagen sie abwesend sind, also nicht zur Verfügung stehen. Zudem können sie für jede Tag-Dienst-Kombination angeben, ob sie den Dienst an diesen Tag bevorzugt (+) oder nur notfalls (-) machen wollen. +## Dienste -Die Eingabe erfolgt über eine CSV-Datei eingabe.csv und eltern.csv +- **F** - Frühstücksdienst (täglich, 1 Person) +- **P** - Putznotdienst (täglich, 1 Person) +- **E** - Essensausgabenotdienst (täglich, 1 Person) +- **K** - Kochen (ca. alle 2 Wochen, 1 Person) +- **A** - Elternabend (nach Bedarf, 2 Personen) -## eingabe.csv -Informationen zu notwendigen Diensten eines Monats und Zeiten/Praeferenzen der Eltern +Die Planung erfolgt für einen Kalendermonat. -1. Spalte: Datum in ISO-Format -2. Spalte: Wochentag (Hilfsinformation) -3. Spalte: Benötigte Dienste als aneinandergereihte Dienstkürzel -Folgende Spalten: Für alle Eltern: Verfügbarkeit und Präferenz: - - x, falls nicht verfügbar - - +, wenn Dienst an dem Tag bevorzugt. - - -, wenn Dienst an dem Tag abgelehnt. -Es können mehrere Präferenzen pro Tag angegeben werden. -1. Zeile header -folgende Zeilen je Tag. +## Eingabedateien +### eingabe.csv -## eltern.csv: -Informationen zur Diestpflicht der Eltern. -Die Dienstpflicht besteht, wenn Eltern Kinder im Kinderladen betreuen lassen. -Der Dienstfaktor entspricht der Anzahl der betreuten Kinder der Eltern. -Wenn Eltern ein Vorsandsamt im Kinderladen übernehmen, werden sie von der Dienstpflicht befreit. +Informationen zu benötigten Diensten und Verfügbarkeit/Präferenzen der Eltern. -1. Spalte Zeitraum Beginn -2. Spalte Zeitraum Ende -3. Spalte Dienstfaktor -4. Spalte ... nächster Zeitraum -1. Zeile Header -folgende Zeilen Eltern +**Format:** +``` +Datum,Wochentag,Dienste,Eltern1,Eltern2,... +2025-01-06,Montag,FPE,F+,x,... +2025-01-07,Dienstag,FPE,P-,F+P+,... +``` +**Spalten:** +1. Datum (ISO-Format: YYYY-MM-DD) +2. Wochentag (zur Information) +3. Benötigte Dienste (z.B. "FPE" für Frühstück, Putzen, Essen) +4-n. Für jeden Elternteil: + - `x` = nicht verfügbar + - `F+` = Frühstücksdienst bevorzugt + - `P-` = Putznotdienst nur notfalls + - Mehrere Präferenzen kombinierbar: `F+P-E+` + - Leer = verfügbar, keine Präferenz -Bei sich überschneidenden Zeiträumen gilt der letzte Eintrag. -An Tagen außerhalb der angegebenen Zeiträume ist der Dienstfaktor 0. +### eltern.csv -Die Datei eltern.csv enthält ggf. mehr Eltern als die Eingabe.csv, -da Kinder dazukommen oder den KiLa verlassen, die eltern.csv aber nur anwächst. +Dienstfaktoren (= Anzahl betreuter Kinder) pro Elternteil und Zeitraum. -## ausgabe.csv -1. Spalte: Datum -2. Spalte: Wochentag -3. Spalte ... Dienste -Zeilen: für jeden Tag die zugeteilten Eltern in den jeweiligen Dienstspalten +**Format:** +``` +Eltern,Beginn,Ende,Faktor,Beginn,Ende,Faktor,... +Müller,2024-09-01,2025-07-31,2 +Schmidt,2024-09-01,2024-12-31,1,2025-01-01,2025-07-31,0 +``` -## vorherige-ausgaben.csv -Hier werden die von früheren Läufen des Programms generierten ausgabe.csv-Datein wiedereingespielt. -Das Format entspricht der ausgabe.csv +**Spalten:** +1. Elternname (Kind-Name zur Identifikation) +2-4. Zeitraum 1: Beginn, Ende, Dienstfaktor +5-7. Zeitraum 2: Beginn, Ende, Dienstfaktor (optional) +... + +**Hinweise:** +- Bei überlappenden Zeiträumen gilt der letzte Eintrag +- Außerhalb definierter Zeiträume: Faktor = 0 (keine Dienstpflicht) +- Faktor = 0 bedeutet: Befreiung (z.B. durch Vorstandsamt) + +### vorherige-ausgaben.csv (optional) + +Frühere Ausgaben des Programms zur Berechnung der Jahres-Fairness. + +**Format:** Wie `ausgabe.csv` (siehe unten). + +**Verwendung:** +- Zu Beginn des Kita-Jahres (September): Keine Datei nötig +- Ab Oktober: Vorherige Ausgaben anhängen für kumulative Fairness +- Im Jahresverlauf sammeln sich die Ausgaben an + +## Ausgabe + +### ausgabe.csv + +Zugeteilte Dienste pro Tag. + +**Format:** +``` +Datum,Wochentag,Frühstücksdienst,Putznotdienst,Essensausgabenotdienst,Kochen,Elternabend +2025-01-06,Montag,Müller,Schmidt,Weber,, +2025-01-07,Dienstag,Weber,Müller,Schmidt,, +``` ## Constraints -Harte Constraints: - C1: Je Eltern und Dienst, Dienst nur einmal die Woche - C2: Je Eltern nur einen Dienst am Tag - C3: Dienste nur verfügbaren Eltern zuteilen +### Harte Constraints (müssen erfüllt sein) -Weiche Constraints: -- F1: Alle Eltern erhalten Dienste im Verhältnis ihres Dienstfaktors (Gesamter vorliegender Zeitraum) -- F2: Alle Eltern erhalten Dienste im Verhältnis ihres aktuellen Dienstfaktors (Aktueller Monat) -- P1: Eltern erhalten bevorzugte Dienste -- P2: Eltern erhalten keine abgelehnten Dienste. +- **C1**: Pro Eltern und Dienst maximal **einmal pro Woche** (Mo-So) +- **C2**: Pro Eltern maximal **ein Dienst pro Tag** +- **C3**: Nur **verfügbare** Eltern einteilen +- **C4**: Alle **benötigten Dienste** müssen besetzt werden -F1 und F2 sind Fairnessconstraints und gelten pro Dienst. -P1 und P2 sind Präferenzconstraints. Sie wiegen schwächer als die Fairnessconstaints. +### Weiche Constraints (werden optimiert) +**Fairness** (nach Priorität): +- **F1 (Global)**: Dienste proportional zum Dienstfaktor über das **ganze Jahr** + - Berücksichtigt historische Dienste aus `vorherige-ausgaben.csv` + - Gewichtung: 40% (zu Jahresbeginn) → 60% (zu Jahresende) -Die vorherige-ausgaben.csv der vergangenen Monate dienen auch als Eingabe. -Sie werden für die Fairnessconstraints verwendet. -Im Laufe eines Kinderladenjahrs sammeln sich die Ausgaben der Monate an. -F2 stellt die Fairness im aktuellen Monat sicher -> lokale Fairness -F1 stellt die Fairness für das gesamte Jahr sicher -> globale Fairness +- **F2 (Lokal)**: Dienste proportional zum Dienstfaktor im **aktuellen Monat** + - Nur aktueller Planungszeitraum + - Gewichtung: 60% (zu Jahresbeginn) → 40% (zu Jahresende) -Wenn z.B. Eltern eine zeitlang nicht verfügbar sind, sollen sie nicht sofort -alle Dienste "nachholen" müssen (lokale Fairness stellt das sicher), -aber im Jahresverlauf die Dienste trotzdem nachholen (globale Fairness stellt das sicher). +- **F3 (Dienstübergreifend)**: Gleiche Gesamtanzahl über alle Diensttypen + - Verhindert Häufung bei einzelnen Eltern -F1 und F2 werden mit Faktoren gewichtet. Zu Beginn des Kinderladenjahrs ist F2 stärker, -zum Ende des Kinderladenjahres F1. +**Präferenzen** (niedrigere Priorität): +- **P1**: Bevorzugte Dienste (`+`) werden bevorzugt zugeteilt +- **P2**: Abgelehnte Dienste (`-`) werden vermieden + +### Fairness-Logik + +**Beispiel:** Familie Müller hat 2 Kinder, Familie Schmidt 1 Kind. + +**Lokale Fairness (F2):** +- Im Januar sollen beide verfügbar sein +- Müller sollte 2× so viele Dienste bekommen wie Schmidt +- Verhindert: Müller bekommt alle Dienste auf einmal + +**Globale Fairness (F1):** +- Müller war im Dezember im Urlaub → 0 Dienste +- Im Januar sollte Müller aufholen +- Über das Jahr: 2:1 Verhältnis wird ausgeglichen + +**Gewichtung im Jahresverlauf:** +- **September-November**: F2 (lokal) stärker → sanftes Einführen +- **Dezember-Mai**: Ausgewogen +- **Juni-Juli**: F1 (global) stärker → Jahresausgleich + +## Ausgabe-Statistiken + +Das Programm zeigt nach der Optimierung: + +1. **Dienste pro Eltern**: Übersicht der zugeteilten Dienste +2. **Dienstfaktoren**: Summe im Planungszeitraum +3. **Verteilungsvergleich**: Soll (lokal/global) vs. Ist mit Abweichungen +4. **Präferenz-Verletzungen**: Wie oft wurden Ablehnungen ignoriert + +## Troubleshooting + +**"Keine optimale Lösung gefunden":** +- Zu viele Eltern nicht verfügbar +- Nicht genug Eltern für alle Dienste +- Widersprüchliche Präferenzen + +**"Unfaire Verteilung":** +- Prüfen Sie die Dienstfaktoren in `eltern.csv` +- Stellen Sie sicher, dass `vorherige-ausgaben.csv` korrekt ist +- Mehr Eltern verfügbar machen diff --git a/STRUKTUR.md b/STRUKTUR.md index 566eb27..e69de29 100644 --- a/STRUKTUR.md +++ b/STRUKTUR.md @@ -1,107 +0,0 @@ -# Projektstruktur Elterndienstplaner - -## Dateien - -### `csv_io.py` - CSV Input/Output Module -**Zweck**: Trennung von Datei-I/O und Business-Logik - -**Klassen**: -- `EingabeParser`: Parst alle CSV-Eingabedateien - - `parse_eingabe_csv()`: Lädt eingabe.csv mit Terminen und Präferenzen - - `parse_eltern_csv()`: Lädt eltern.csv mit Dienstfaktoren - - `parse_vorherige_ausgaben_csv()`: Lädt vorherige-ausgaben.csv für Fairness - -- `AusgabeWriter`: Schreibt Ergebnisse in CSV - - `schreibe_ausgabe_csv()`: Schreibt Lösung in ausgabe.csv - -**Vorteile**: -- ✅ Testbar: CSV-Parsing kann isoliert getestet werden -- ✅ Wiederverwendbar: Andere Formate (JSON, Excel) leicht hinzufügbar -- ✅ Klare Verantwortlichkeiten: I/O getrennt von Optimierung - -### `elterndienstplaner.py` - Hauptprogramm -**Zweck**: Business-Logik und Optimierung - -**Klassen**: -- `Dienst`: Datenmodell für Diensttypen -- `Elterndienstplaner`: Hauptklasse mit Optimierungslogik - - Fairness-Berechnungen (global/lokal) - - Optimierungsmodell-Erstellung (modular aufgeteilt) - - Statistiken - -**Constraint-Funktionen** (modular aufgeteilt): -- `_erstelle_entscheidungsvariablen()`: Erstellt binäre Variablen -- `_add_constraint_ein_dienst_pro_woche()`: C1 - Max 1 Dienst pro Woche -- `_add_constraint_ein_dienst_pro_tag()`: C2 - Max 1 Dienst pro Tag -- `_add_constraint_verfuegbarkeit()`: C3 - Nur verfügbare Eltern -- `_add_constraint_dienst_bedarf()`: C4 - Alle Dienste müssen besetzt werden -- `_add_fairness_constraints()`: F1 & F2 - Erstellt Variablen und Constraints für Fairness -- `_berechne_fairness_gewichte()`: Zeitabhängige Gewichtung (Sep-Jul) -- `_erstelle_zielfunktion()`: Zielfunktion mit Fairness & Präferenzen - -**Abhängigkeiten**: -- Importiert `csv_io` für Datei-Operationen -- Verwendet `pulp` für lineare Optimierung - -## Verbesserungen durch Refactoring - -### Vorher (Monolith) -``` -elterndienstplaner.py (700+ Zeilen) -├── Dienst Klasse -├── CSV Parsing (150+ Zeilen) -├── Fairness-Berechnung -├── Optimierung (200+ Zeilen inline) -└── CSV Schreiben -``` - -### Nachher (Modular) -``` -csv_io.py (220 Zeilen) -├── EingabeParser -└── AusgabeWriter - -elterndienstplaner.py (500 Zeilen) -├── Dienst Klasse -├── Fairness-Berechnung -│ ├── berechne_faire_zielverteilung_global() -│ └── berechne_faire_zielverteilung_lokal() -├── Optimierung (modular) -│ ├── erstelle_optimierungsmodell() (30 Zeilen - übersichtlich!) -│ ├── _erstelle_entscheidungsvariablen() -│ ├── _add_constraint_ein_dienst_pro_woche() -│ ├── _add_constraint_ein_dienst_pro_tag() -│ ├── _add_constraint_verfuegbarkeit() -│ ├── _add_constraint_dienst_bedarf() -│ ├── _add_fairness_constraints() -│ └── _erstelle_zielfunktion() -└── Statistiken -``` - -## Nächste Schritte (Optional) - -### Phase 3: Fairness-Modul (optional) -``` -fairness.py -├── FairnessBerechner -│ ├── berechne_global() -│ └── berechne_lokal() -```## Verwendung - -```python -from csv_io import EingabeParser, AusgabeWriter - -# Daten laden -eltern, tage, dienste, ... = EingabeParser.parse_eingabe_csv("eingabe.csv", lookup_fn) - -# Ergebnis schreiben -AusgabeWriter.schreibe_ausgabe_csv("ausgabe.csv", lösung, tage, dienste) -``` - -## Vorteile der aktuellen Struktur - -1. **Separation of Concerns**: I/O getrennt von Business-Logik -2. **Testbarkeit**: Module können unabhängig getestet werden -3. **Wartbarkeit**: Änderungen an CSV-Format betreffen nur `csv_io.py` -4. **Erweiterbarkeit**: Neue Dateiformate können leicht hinzugefügt werden -5. **Lesbarkeit**: Kürzere, fokussiertere Dateien diff --git a/ausgabe.py b/ausgabe.py new file mode 100644 index 0000000..6cd8488 --- /dev/null +++ b/ausgabe.py @@ -0,0 +1,251 @@ +#!/usr/bin/env python3 +""" +Ausgabe-Modul für Elterndienstplaner +Visualisierung und Export der Ergebnisse +""" + +from datetime import date +from collections import defaultdict +from typing import Dict, List, DefaultDict + +from datenmodell import ElterndienstplanerDaten, Dienst, Eltern, Zielverteilung +from csv_io import AusgabeWriter + + +class ElterndienstAusgabe: + """Ausgabe und Visualisierung der Optimierungsergebnisse""" + + def __init__(self, daten: ElterndienstplanerDaten) -> None: + self.daten = daten + # Zwischenergebnisse aus der Optimierung (über Observer-Pattern gesetzt) + self.ziel_lokal: Zielverteilung = None + self.ziel_global: Zielverteilung = None + + def setze_zielverteilungen( + self, + ziel_lokal: Zielverteilung, + ziel_global: Zielverteilung + ) -> None: + """Setzt die Zielverteilungen (Observer-Callback)""" + self.ziel_lokal = ziel_lokal + self.ziel_global = ziel_global + + def schreibe_ausgabe_csv(self, datei: str, lösung: Dict[date, Dict[Dienst, List[Eltern]]]) -> None: + """Schreibt die Lösung in die ausgabe.csv""" + AusgabeWriter.schreibe_ausgabe_csv(datei, lösung, self.daten.planungszeitraum, self.daten.dienste) + + def drucke_statistiken(self, lösung: Dict[date, Dict[Dienst, List[Eltern]]]) -> None: + """Druckt Statistiken zur Lösung""" + print("\n" + "="*50) + print("STATISTIKEN") + print("="*50) + + # Dienste pro Eltern zählen + dienste_pro_eltern = defaultdict(lambda: defaultdict(int)) + for tag, tag_dienste in lösung.items(): + for dienst, eltern_liste in tag_dienste.items(): + for eltern in eltern_liste: + dienste_pro_eltern[eltern][dienst] += 1 + + # Gesamtübersicht + print("\nDienste pro Eltern:") + for eltern in sorted(self.daten.eltern): + gesamt = sum(dienste_pro_eltern[eltern].values()) + dienste_detail = ', '.join(f"{dienst.kuerzel}:{dienste_pro_eltern[eltern][dienst]}" + for dienst in self.daten.dienste if dienste_pro_eltern[eltern][dienst] > 0) + print(f" {eltern:15} {gesamt:3d} ({dienste_detail})") + + # Dienstfaktor-Analyse + print(f"\nDienstfaktoren im Planungszeitraum:") + for eltern in sorted(self.daten.eltern): + faktor_summe = sum(self.daten.dienstfaktoren[eltern][tag] for tag in self.daten.planungszeitraum) + print(f" {eltern:15} {faktor_summe:.1f}") + + def visualisiere_praeferenz_verletzungen( + self, + lösung: Dict[date, Dict[Dienst, List[Eltern]]] + ) -> None: + """Visualisiert verletzte Präferenzen als Tabelle + + Args: + lösung: Die tatsächliche Lösung nach Optimierung + """ + print("\n" + "="*110) + print("PRÄFERENZ-VERLETZUNGEN") + print("="*110) + + # Sammle alle zugeteilten Dienste pro Eltern + zugeteilte_dienste = defaultdict(lambda: defaultdict(list)) # eltern -> dienst -> [dates] + for tag, tag_dienste in lösung.items(): + for dienst, eltern_liste in tag_dienste.items(): + for eltern in eltern_liste: + zugeteilte_dienste[eltern][dienst].append(tag) + + # Sammle Präferenzen strukturiert + # praeferenzen_pro_eltern_dienst[eltern][dienst] = {datum: präf_wert} + praeferenzen_pro_eltern_dienst = defaultdict(lambda: defaultdict(dict)) + for (eltern, tag, dienst), präf in self.daten.präferenzen.items(): + praeferenzen_pro_eltern_dienst[eltern][dienst][tag] = präf + + # Berechne Verletzungen + verletzungen = defaultdict(lambda: defaultdict(lambda: {'negativ': 0, 'positiv_nicht_erfuellt': 0})) + + for eltern in sorted(self.daten.eltern): + for dienst in self.daten.dienste: + zugeteilte_tage = zugeteilte_dienste[eltern][dienst] + praeferenzen_dienst = praeferenzen_pro_eltern_dienst[eltern][dienst] + + if not zugeteilte_tage: + continue # Keine Dienste zugeteilt + + # a) Negative Präferenzen die verletzt wurden + for tag in zugeteilte_tage: + if tag in praeferenzen_dienst and praeferenzen_dienst[tag] == -1: + verletzungen[eltern][dienst]['negativ'] += 1 + + # b) Positive Präferenzen nicht erfüllt (Dienst an nicht-präferiertem Tag) + # Sammle alle Tage mit positiver Präferenz für diesen Dienst + positive_praef_tage = {tag for tag, präf in praeferenzen_dienst.items() if präf == 1} + + if positive_praef_tage: # Es gibt positive Präferenzen + # Prüfe ob ALLE zugeteilten Dienste an nicht-präferierten Tagen sind + for tag in zugeteilte_tage: + if tag not in positive_praef_tage: + # Dienst wurde an nicht-präferiertem Tag zugeteilt + verletzungen[eltern][dienst]['positiv_nicht_erfuellt'] += 1 + + # Tabelle ausgeben + print(f"\n{'Eltern':<20} ", end='') + for dienst in self.daten.dienste: + print(f"{dienst.kuerzel:>12}", end='') + print() + print(f"{'':20} ", end='') + for dienst in self.daten.dienste: + print(f"{'neg, pos':>12}", end='') + print() + print("-" * (20 + 12 * len(self.daten.dienste))) + + gesamt_negativ = defaultdict(int) + gesamt_positiv = defaultdict(int) + + for eltern in sorted(self.daten.eltern): + print(f"{eltern:<20} ", end='') + for dienst in self.daten.dienste: + neg = verletzungen[eltern][dienst]['negativ'] + pos = verletzungen[eltern][dienst]['positiv_nicht_erfuellt'] + + gesamt_negativ[dienst] += neg + gesamt_positiv[dienst] += pos + + # Farbcodierung + farbe = "" + reset = "" + if neg > 0 or pos > 0: + farbe = "\033[91m" if neg > 0 else "\033[93m" # Rot für negativ, Gelb für positiv + reset = "\033[0m" + + print(f"{farbe}{neg:>3}, {pos:>3}{reset:>6}", end='') + print() + + # Summenzeile + print("-" * (20 + 12 * len(self.daten.dienste))) + print(f"{'SUMME':<20} ", end='') + for dienst in self.daten.dienste: + neg = gesamt_negativ[dienst] + pos = gesamt_positiv[dienst] + + farbe = "" + reset = "" + if neg > 0 or pos > 0: + farbe = "\033[91m" if neg > 0 else "\033[93m" + reset = "\033[0m" + + print(f"{farbe}{neg:>3}, {pos:>3}{reset:>6}", end='') + print() + + print("\nLegende:") + print(" neg = Anzahl negativer Präferenzen (abgelehnte Tage), die verletzt wurden") + print(" pos = Anzahl Dienste an nicht-präferierten Tagen (obwohl präferierte Tage angegeben waren)") + print(" \033[91mRot\033[0m = Negative Präferenz verletzt") + print(" \033[93mGelb\033[0m = Positive Präferenz nicht erfüllt") + + def visualisiere_verteilungen( + self, + lösung: Dict[date, Dict[Dienst, List[Eltern]]] + ) -> None: + """Visualisiert die Verteilungen als Tabelle zum Vergleich + + Args: + lösung: Die tatsächliche Lösung nach Optimierung + """ + if self.ziel_lokal is None or self.ziel_global is None: + print("FEHLER: Zielverteilungen wurden nicht gesetzt!") + return + + # Tatsächliche Dienste zählen + tatsaechlich = defaultdict(lambda: defaultdict(int)) + for tag_dienste in lösung.values(): + for dienst, eltern_liste in tag_dienste.items(): + for eltern in eltern_liste: + tatsaechlich[eltern][dienst] += 1 + + print("\n" + "="*110) + print("VERTEILUNGSVERGLEICH: SOLL vs. IST") + print("="*110) + + for dienst in self.daten.dienste: + print(f"\n{dienst.name} ({dienst.kuerzel}):") + print(f"{'Eltern':<20} {'Ziel Global':>12} {'Ziel Lokal':>12} {'Tatsächlich':>12} " + f"{'Δ Global':>12} {'Δ Lokal':>12}") + print("-" * 110) + + for eltern in sorted(self.daten.eltern): + z_global = self.ziel_global[eltern][dienst] + z_lokal = self.ziel_lokal[eltern][dienst] + ist = tatsaechlich[eltern][dienst] + delta_global = ist - z_global + delta_lokal = ist - z_lokal + + # Farbcodierung für Abweichungen (ANSI-Codes) + farbe_global = "" + farbe_lokal = "" + reset = "" + + if abs(delta_global) > 0.5: + farbe_global = "\033[93m" if abs(delta_global) <= 1.0 else "\033[91m" # Gelb oder Rot + reset = "\033[0m" + + if abs(delta_lokal) > 0.5: + farbe_lokal = "\033[93m" if abs(delta_lokal) <= 1.0 else "\033[91m" + reset = "\033[0m" + + print(f"{eltern:<20} {z_global:>12.2f} {z_lokal:>12.2f} {ist:>12} " + f"{farbe_global}{delta_global:>+12.2f}{reset} {farbe_lokal}{delta_lokal:>+12.2f}{reset}") + + # Summen + summe_z_global = sum(self.ziel_global[e][dienst] for e in self.daten.eltern) + summe_z_lokal = sum(self.ziel_lokal[e][dienst] for e in self.daten.eltern) + summe_ist = sum(tatsaechlich[e][dienst] for e in self.daten.eltern) + + print("-" * 110) + print(f"{'SUMME':<20} {summe_z_global:>12.2f} {summe_z_lokal:>12.2f} {summe_ist:>12} " + f"{summe_ist - summe_z_global:>+12.2f} {summe_ist - summe_z_lokal:>+12.2f}") + + # Gesamtstatistik + print("\n" + "="*110) + print("ZUSAMMENFASSUNG") + print("="*110) + + # Maximale Abweichungen finden + max_abw_global = 0 + max_abw_lokal = 0 + + for eltern in self.daten.eltern: + for dienst in self.daten.dienste: + ist = tatsaechlich[eltern][dienst] + max_abw_global = max(max_abw_global, abs(ist - self.ziel_global[eltern][dienst])) + max_abw_lokal = max(max_abw_lokal, abs(ist - self.ziel_lokal[eltern][dienst])) + + print(f"Maximale Abweichung von Global-Ziel: {max_abw_global:.2f} Dienste") + print(f"Maximale Abweichung von Lokal-Ziel: {max_abw_lokal:.2f} Dienste") + print("\nLegende: Δ = Tatsächlich - Ziel (positiv = mehr als Ziel, negativ = weniger als Ziel)") diff --git a/datenmodell.py b/datenmodell.py new file mode 100644 index 0000000..3a49f8f --- /dev/null +++ b/datenmodell.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +""" +Datenmodell für Elterndienstplaner +Enthält alle Daten und deren Laden sowie gemeinsame Type Aliases +""" + +from datetime import date +from collections import defaultdict +from typing import Dict, List, Tuple, DefaultDict, Optional, TypeAlias +import pulp + +from csv_io import EingabeParser + + +class Dienst: + """Repräsentiert einen Diensttyp mit allen seinen Eigenschaften""" + + def __init__(self, kuerzel: str, name: str, personen_anzahl: int = 1) -> None: + self.kuerzel: str = kuerzel + self.name: str = name + self.personen_anzahl: int = personen_anzahl + + def __str__(self) -> str: + return f"{self.kuerzel} ({self.name}): {self.personen_anzahl} Person(en)" + + def __repr__(self) -> str: + return f"Dienst('{self.kuerzel}', '{self.name}', {self.personen_anzahl})" + + def braucht_mehrere_personen(self) -> bool: + """Gibt True zurück, wenn mehr als eine Person benötigt wird""" + return self.personen_anzahl > 1 + +# Eltern: Ihnen koennen Dienste zugewiesen werden +Eltern: TypeAlias = str + +# Verteilung von Diensten auf Eltern. +# - Zielsumme der Dienste ueber den Planungszeitraum +# - Kann nicht-ganzzahlig und negativ sein +Zielverteilung: TypeAlias = DefaultDict[Eltern, DefaultDict[Dienst, float]] + +# Entscheidungsvariablen des Optimierungsproblems +# Variable: Eltern wird am Datum Dienst zugewiesen +Entscheidungsvariablen: TypeAlias = Dict[Tuple[Eltern, date, Dienst], pulp.LpVariable] + + +class ElterndienstplanerDaten: + """Datenmodell für den Elterndienstplaner""" + + def __init__(self) -> None: + # Dienste als Liste definieren + self.dienste: List[Dienst] = [ + Dienst('F', 'Frühstücksdienst', 1), + Dienst('P', 'Putznotdienst', 1), + Dienst('E', 'Essensausgabenotdienst', 1), + Dienst('K', 'Kochen', 1), + Dienst('A', 'Elternabend', 2) + ] + + # Datenstrukturen + self.planungszeitraum: List[date] = [] + self.eltern: List[Eltern] = [] + self.benoetigte_dienste: Dict[date, List[Dienst]] = {} + self.verfügbarkeit: Dict[Tuple[Eltern, date], bool] = {} + self.präferenzen: Dict[Tuple[Eltern, date, Dienst], int] = {} + + # dienstfaktoren[eltern][tag] = faktor. + # Wenn es eltern nicht gibt -> keyerror + # Wenn es tag nicht gibt -> default 0.0 + self.dienstfaktoren: Dict[Eltern, DefaultDict[date, float]] = {} + self.historische_dienste: List[Tuple[date, Eltern, Dienst]] = [] + + def get_dienst(self, kuerzel: str) -> Optional[Dienst]: + """Gibt das Dienst-Objekt für ein Kürzel zurück""" + for dienst in self.dienste: + if dienst.kuerzel == kuerzel: + return dienst + return None + + def add_dienst(self, kuerzel: str, name: str, personen_anzahl: int = 1) -> Dienst: + """Fügt einen neuen Dienst hinzu""" + dienst = Dienst(kuerzel, name, personen_anzahl) + self.dienste.append(dienst) + return dienst + + def print_dienste_info(self) -> None: + """Druckt Informationen über alle konfigurierten Dienste""" + print("Konfigurierte Dienste:") + for dienst in self.dienste: + print(f" {dienst}") + + def lade_daten( + self, + eingabe_datei: str, + eltern_datei: str, + vorherige_datei: Optional[str] = None + ) -> None: + """Lädt alle benötigten CSV-Dateien + + Args: + eingabe_datei: Pfad zur eingabe.csv mit Terminen und Präferenzen + eltern_datei: Pfad zur eltern.csv mit Dienstfaktoren + vorherige_datei: Optionaler Pfad zur vorherige-ausgaben.csv für Fairness-Constraints + """ + # Eingabe CSV: Termine, Präferenzen, Verfügbarkeit + self.eltern, self.planungszeitraum, self.benoetigte_dienste, self.verfügbarkeit, self.präferenzen = \ + EingabeParser.parse_eingabe_csv(eingabe_datei, self.get_dienst) + + # Eltern CSV: Dienstfaktoren + self.dienstfaktoren = EingabeParser.parse_eltern_csv(eltern_datei) + + # Vorherige Ausgaben CSV (optional): Historische Dienste für Fairness + if vorherige_datei: + self.historische_dienste = \ + EingabeParser.parse_vorherige_ausgaben_csv(vorherige_datei, self.eltern, self.dienste) diff --git a/elterndienstplaner.py b/elterndienstplaner.py index 06bd519..6429b90 100755 --- a/elterndienstplaner.py +++ b/elterndienstplaner.py @@ -12,92 +12,18 @@ from datetime import timedelta, date from collections import defaultdict from typing import Dict, List, Tuple, DefaultDict, Optional -from csv_io import EingabeParser, AusgabeWriter - - -class Dienst: - """Repräsentiert einen Diensttyp mit allen seinen Eigenschaften""" - - def __init__(self, kuerzel: str, name: str, personen_anzahl: int = 1) -> None: - self.kuerzel: str = kuerzel - self.name: str = name - self.personen_anzahl: int = personen_anzahl - - def __str__(self) -> str: - return f"{self.kuerzel} ({self.name}): {self.personen_anzahl} Person(en)" - - def __repr__(self) -> str: - return f"Dienst('{self.kuerzel}', '{self.name}', {self.personen_anzahl})" - - def braucht_mehrere_personen(self) -> bool: - """Gibt True zurück, wenn mehr als eine Person benötigt wird""" - return self.personen_anzahl > 1 +from datenmodell import ElterndienstplanerDaten, Dienst, Eltern, Zielverteilung, Entscheidungsvariablen +from ausgabe import ElterndienstAusgabe class Elterndienstplaner: - def __init__(self) -> None: - # Dienste als Liste definieren - self.dienste: List[Dienst] = [ - Dienst('F', 'Frühstücksdienst', 1), - Dienst('P', 'Putznotdienst', 1), - Dienst('E', 'Essensausgabenotdienst', 1), - Dienst('K', 'Kochen', 1), - Dienst('A', 'Elternabend', 2) - ] + """Optimierungs-Engine für Elterndienstplanung""" - # Datenstrukturen - self.planungszeitraum: List[date] = [] - self.eltern: List[str] = [] - self.benoetigte_dienste: Dict[date, List[Dienst]] = {} - self.verfügbarkeit: Dict[Tuple[str, date], bool] = {} - self.präferenzen: Dict[Tuple[str, date, Dienst], int] = {} + def __init__(self, daten: ElterndienstplanerDaten, ausgabe: ElterndienstAusgabe) -> None: + self.daten = daten + self.ausgabe = ausgabe - # dienstfaktoren[eltern][tag] = faktor. - # Wenn es eltern nicht gibt -> keyerror - # Wenn es tag nicht gibt -> default 0.0 - self.dienstfaktoren: Dict[str, DefaultDict[date, float]] = {} - self.historische_dienste: List[Tuple[date, str, Dienst]] = [] - - def get_dienst(self, kuerzel: str) -> Optional[Dienst]: - """Gibt das Dienst-Objekt für ein Kürzel zurück""" - for dienst in self.dienste: - if dienst.kuerzel == kuerzel: - return dienst - return None - - def add_dienst(self, kuerzel: str, name: str, personen_anzahl: int = 1) -> Dienst: - """Fügt einen neuen Dienst hinzu""" - dienst = Dienst(kuerzel, name, personen_anzahl) - self.dienste.append(dienst) - return dienst - - def print_dienste_info(self) -> None: - """Druckt Informationen über alle konfigurierten Dienste""" - print("Konfigurierte Dienste:") - for dienst in self.dienste: - print(f" {dienst}") - - def lade_eingabe_csv(self, datei: str) -> None: - """Lädt die eingabe.csv mit Terminen und Präferenzen""" - self.eltern, self.planungszeitraum, self.benoetigte_dienste, self.verfügbarkeit, self.präferenzen = \ - EingabeParser.parse_eingabe_csv(datei, self.get_dienst) - - def lade_eltern_csv(self, datei: str) -> None: - """Lädt die eltern.csv mit Dienstfaktoren""" - self.dienstfaktoren = EingabeParser.parse_eltern_csv(datei) - - def lade_vorherige_ausgaben_csv(self, datei: str) -> None: - """Lädt vorherige-ausgaben.csv für Fairness-Constraints""" - self.historische_dienste = \ - EingabeParser.parse_vorherige_ausgaben_csv(datei, self.eltern, self.dienste) - - def berechne_dienstfaktor_an_datum(self, eltern: str, datum: date) -> float: - """Berechnet den Dienstfaktor eines Elternteils an einem bestimmten Datum""" - if eltern not in self.dienstfaktoren: - return 0 - return self.dienstfaktoren[eltern][datum] # DefaultDict gibt 0 zurück für unbekannte Tage - - def berechne_faire_zielverteilung_global(self) -> DefaultDict[str, DefaultDict[Dienst, float]]: + def berechne_faire_zielverteilung_global(self) -> Zielverteilung: """Berechnet die faire Zielanzahl von Diensten für den Planungszeitraum basierend auf globaler Fairness (Historie + aktueller Planungszeitraum). @@ -105,21 +31,20 @@ class Elterndienstplaner: korrigiert um bereits geleistete Dienste. Kann negativ sein, wenn bereits mehr Dienste geleistet wurden als fair wäre.""" - ziel_dienste: DefaultDict[str, DefaultDict[Dienst, float]] = \ - defaultdict(lambda: defaultdict(float)) + ziel_dienste: Zielverteilung = defaultdict(lambda: defaultdict(float)) print("\nBerechne faire Zielverteilung basierend auf historischen Daten...") # Historische Dienste nach Datum gruppieren - historische_tage = set(datum for datum, _, _ in self.historische_dienste) if self.historische_dienste else set() - print(f" Analysiere {len(historische_tage)} historische Tage mit {len(self.historische_dienste)} Diensten") + historische_tage = set(datum for datum, _, _ in self.daten.historische_dienste) if self.daten.historische_dienste else set() + print(f" Analysiere {len(historische_tage)} historische Tage mit {len(self.daten.historische_dienste)} Diensten") - for dienst in self.dienste: + for dienst in self.daten.dienste: print(f" Verarbeite Dienst {dienst.kuerzel}...") # 1. HISTORISCHE PERIODE: Faire Umverteilung der tatsächlich geleisteten Dienste historische_dienste_dieses_typs = [ - (datum, eltern) for datum, eltern, d in self.historische_dienste + (datum, eltern) for datum, eltern, d in self.daten.historische_dienste if d == dienst ] @@ -137,28 +62,28 @@ class Elterndienstplaner: # Dienstfaktoren aller Eltern für diesen historischen Tag berechnen gesamt_dienstfaktor_tag = 0 - for eltern in self.eltern: - gesamt_dienstfaktor_tag += self.dienstfaktoren[eltern][tag] + for eltern in self.daten.eltern: + gesamt_dienstfaktor_tag += self.daten.dienstfaktoren[eltern][tag] # Faire Umverteilung der an diesem Tag geleisteten Dienste if gesamt_dienstfaktor_tag > 0: - for eltern in self.eltern: - if self.dienstfaktoren[eltern][tag] > 0: - anteil = self.dienstfaktoren[eltern][tag] / gesamt_dienstfaktor_tag + for eltern in self.daten.eltern: + if self.daten.dienstfaktoren[eltern][tag] > 0: + anteil = self.daten.dienstfaktoren[eltern][tag] / gesamt_dienstfaktor_tag faire_zuteilung = anteil * anzahl_dienste ziel_dienste[eltern][dienst] += faire_zuteilung if faire_zuteilung > 0.01: # Debug nur für relevante Werte - print(f" {tag}: {eltern} Faktor={self.dienstfaktoren[eltern][tag]} " + print(f" {tag}: {eltern} Faktor={self.daten.dienstfaktoren[eltern][tag]} " f"-> {faire_zuteilung:.2f} von {anzahl_dienste} Diensten") # 2. AKTUELLER PLANUNGSZEITRAUM: Faire Verteilung der benötigten Dienste (tageweise wie bei historischen Diensten) benoetigte_dienste_planungszeitraum = 0 # Für jeden Tag im aktuellen Planungszeitraum faire Umverteilung berechnen - for tag in self.planungszeitraum: + for tag in self.daten.planungszeitraum: # Prüfe ob an diesem Tag der Dienst benötigt wird - if dienst not in self.benoetigte_dienste.get(tag, []): + if dienst not in self.daten.benoetigte_dienste.get(tag, []): continue benoetigte_dienste_planungszeitraum += dienst.personen_anzahl @@ -167,43 +92,42 @@ class Elterndienstplaner: dienstfaktoren = {} gesamt_dienstfaktor_tag = 0 - for eltern in self.eltern: - faktor = self.dienstfaktoren[eltern][tag] + for eltern in self.daten.eltern: + faktor = self.daten.dienstfaktoren[eltern][tag] dienstfaktoren[eltern] = faktor gesamt_dienstfaktor_tag += faktor # Faire Umverteilung der an diesem Tag benötigten Dienste if gesamt_dienstfaktor_tag > 0: - for eltern in self.eltern: + for eltern in self.daten.eltern: anteil = dienstfaktoren[eltern] / gesamt_dienstfaktor_tag faire_zuteilung = anteil * dienst.personen_anzahl ziel_dienste[eltern][dienst] += faire_zuteilung # 3. ABZUG DER BEREITS GELEISTETEN DIENSTE # Ziehe die tatsächlich geleisteten Dienste ab, um das Ziel für den Planungszeitraum zu erhalten - for eltern in self.eltern: + for eltern in self.daten.eltern: # Berechne vorherige Dienste on-the-fly aus historischen Diensten vorherige_anzahl = sum( - 1 for _, hist_eltern, hist_dienst in self.historische_dienste + 1 for _, hist_eltern, hist_dienst in self.daten.historische_dienste if hist_eltern == eltern and hist_dienst == dienst ) ziel_dienste[eltern][dienst] -= vorherige_anzahl return ziel_dienste - def berechne_faire_zielverteilung_lokal(self) -> DefaultDict[str, DefaultDict[Dienst, float]]: + def berechne_faire_zielverteilung_lokal(self) -> Zielverteilung: """Berechnet die lokale faire Zielanzahl von Diensten pro Eltern-Dienst-Kombination basierend auf Dienstfaktoren und benötigten Diensten im aktuellen Planungszeitraum""" - ziel_dienste_lokal: DefaultDict[str, DefaultDict[Dienst, float]] = \ - defaultdict(lambda: defaultdict(float)) + ziel_dienste_lokal: Zielverteilung = defaultdict(lambda: defaultdict(float)) print("\nBerechne lokale faire Zielverteilung für aktuellen Planungszeitraum...") # Gesamtdienstfaktor für aktuellen Planungszeitraum berechnen summe_dienstfaktor_planungszeitraum_alle_eltern = sum( - sum(self.dienstfaktoren[e][tag] for tag in self.planungszeitraum) - for e in self.eltern + sum(self.daten.dienstfaktoren[e][tag] for tag in self.daten.planungszeitraum) + for e in self.daten.eltern ) if summe_dienstfaktor_planungszeitraum_alle_eltern == 0: @@ -211,11 +135,11 @@ class Elterndienstplaner: return ziel_dienste_lokal # Für jeden Dienst die lokale faire Verteilung berechnen - for dienst in self.dienste: + for dienst in self.daten.dienste: # Anzahl benötigter Dienste im aktuellen Planungszeitraum benoetigte_dienste_planungszeitraum = sum( - 1 for tag in self.planungszeitraum - if dienst in self.benoetigte_dienste.get(tag, []) + 1 for tag in self.daten.planungszeitraum + if dienst in self.daten.benoetigte_dienste.get(tag, []) ) # Multipliziere mit Anzahl benötigter Personen pro Dienst benoetigte_dienste_planungszeitraum *= dienst.personen_anzahl @@ -223,10 +147,10 @@ class Elterndienstplaner: if benoetigte_dienste_planungszeitraum > 0: print(f" {dienst.kuerzel}: {benoetigte_dienste_planungszeitraum} Dienste benötigt") - for eltern in self.eltern: + for eltern in self.daten.eltern: # Dienstfaktor für diesen Elternteil im aktuellen Planungszeitraum summe_dienstfaktor_planungszeitraum = sum( - self.dienstfaktoren[eltern][tag] for tag in self.planungszeitraum + self.daten.dienstfaktoren[eltern][tag] for tag in self.daten.planungszeitraum ) if summe_dienstfaktor_planungszeitraum > 0: @@ -236,13 +160,13 @@ class Elterndienstplaner: return ziel_dienste_lokal - def _erstelle_entscheidungsvariablen(self) -> Dict[Tuple[str, date, Dienst], pulp.LpVariable]: + def _erstelle_entscheidungsvariablen(self) -> Entscheidungsvariablen: """Erstellt die binären Entscheidungsvariablen x[eltern, tag, dienst]""" - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable] = {} - for eltern in self.eltern: - for tag in self.planungszeitraum: - for dienst in self.dienste: - if dienst in self.benoetigte_dienste.get(tag, []): + x: Entscheidungsvariablen = {} + for eltern in self.daten.eltern: + for tag in self.daten.planungszeitraum: + for dienst in self.daten.dienste: + if dienst in self.daten.benoetigte_dienste.get(tag, []): x[eltern, tag, dienst] = pulp.LpVariable( f"x_{eltern.replace(' ', '_')}_{tag}_{dienst.kuerzel}", cat='Binary' @@ -252,35 +176,35 @@ class Elterndienstplaner: def _add_constraint_ein_dienst_pro_woche( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable] + x: Entscheidungsvariablen ) -> None: """C1: Je Eltern und Dienst nur einmal die Woche (Woche = Montag bis Sonntag)""" - erster_tag = self.planungszeitraum[0] + erster_tag = self.daten.planungszeitraum[0] # weekday(): 0=Montag, 6=Sonntag # Finde Montag am oder vor dem ersten Planungstag (für historische Dienste) woche_start = erster_tag - timedelta(days=erster_tag.weekday()) woche_nr = 0 - letzter_tag = self.planungszeitraum[-1] + letzter_tag = self.daten.planungszeitraum[-1] while woche_start <= letzter_tag: woche_ende = woche_start + timedelta(days=6) # Sonntag - for eltern in self.eltern: - for dienst in self.dienste: + for eltern in self.daten.eltern: + for dienst in self.daten.dienste: woche_vars = [] # Zähle historische Dienste in dieser Woche (VOR dem Planungszeitraum) historische_dienste_in_woche = 0 if woche_start < erster_tag: - for hist_datum, hist_eltern, hist_dienst in self.historische_dienste: + for hist_datum, hist_eltern, hist_dienst in self.daten.historische_dienste: if (hist_eltern == eltern and hist_dienst == dienst and woche_start <= hist_datum < erster_tag): historische_dienste_in_woche += 1 # Sammle Variablen für Planungszeitraum in dieser Woche - for tag in self.planungszeitraum: + for tag in self.daten.planungszeitraum: if woche_start <= tag <= woche_ende: if (eltern, tag, dienst) in x: woche_vars.append(x[eltern, tag, dienst]) @@ -296,13 +220,13 @@ class Elterndienstplaner: def _add_constraint_ein_dienst_pro_tag( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable] + x: Entscheidungsvariablen ) -> None: """C2: Je Eltern nur einen Dienst am Tag""" - for eltern in self.eltern: - for tag in self.planungszeitraum: + for eltern in self.daten.eltern: + for tag in self.daten.planungszeitraum: tag_vars = [] - for dienst in self.dienste: + for dienst in self.daten.dienste: if (eltern, tag, dienst) in x: tag_vars.append(x[eltern, tag, dienst]) @@ -312,13 +236,13 @@ class Elterndienstplaner: def _add_constraint_verfuegbarkeit( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable] + x: Entscheidungsvariablen ) -> None: """C3: Dienste nur verfügbaren Eltern zuteilen""" - for eltern in self.eltern: - for tag in self.planungszeitraum: - if not self.verfügbarkeit.get((eltern, tag), True): - for dienst in self.dienste: + for eltern in self.daten.eltern: + for tag in self.daten.planungszeitraum: + if not self.daten.verfügbarkeit.get((eltern, tag), True): + for dienst in self.daten.dienste: if (eltern, tag, dienst) in x: prob += x[eltern, tag, dienst] == 0, \ f"C3_{eltern.replace(' ', '_')}_{tag}_{dienst.kuerzel}" @@ -326,16 +250,16 @@ class Elterndienstplaner: def _add_constraint_dienst_bedarf( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable] + x: Entscheidungsvariablen ) -> None: """C4: Alle benötigten Dienste müssen zugeteilt werden""" - for tag in self.planungszeitraum: - for dienst in self.benoetigte_dienste.get(tag, []): + for tag in self.daten.planungszeitraum: + for dienst in self.daten.benoetigte_dienste.get(tag, []): dienst_vars = [] - for eltern in self.eltern: + for eltern in self.daten.eltern: if (eltern, tag, dienst) in x: # Prüfe ob Eltern verfügbar - if self.verfügbarkeit.get((eltern, tag), True): + if self.daten.verfügbarkeit.get((eltern, tag), True): dienst_vars.append(x[eltern, tag, dienst]) if dienst_vars: @@ -347,8 +271,8 @@ class Elterndienstplaner: def _add_fairness_constraints( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable], - ziel_dienste: DefaultDict[str, DefaultDict[Dienst, float]], + x: Entscheidungsvariablen, + ziel_dienste: Zielverteilung, constraint_prefix: str ) -> Dict: """Erstellt Fairness-Variablen und fügt Fairness-Constraints hinzu @@ -365,19 +289,19 @@ class Elterndienstplaner: # Hilfsvariablen für Fairness-Abweichungen erstellen fairness_abweichung = {} - for eltern in self.eltern: - for dienst in self.dienste: + for eltern in self.daten.eltern: + for dienst in self.daten.dienste: fairness_abweichung[eltern, dienst] = pulp.LpVariable( f"fair_{constraint_prefix}_{eltern.replace(' ', '_')}_{dienst.kuerzel}", lowBound=0) # Fairness-Constraints hinzufügen - for eltern in self.eltern: - for dienst in self.dienste: + for eltern in self.daten.eltern: + for dienst in self.daten.dienste: # Tatsächliche Dienste im aktuellen Planungszeitraum zugeteilte_dienste_planungszeitraum = pulp.lpSum( x[eltern, tag, dienst] - for tag in self.planungszeitraum + for tag in self.daten.planungszeitraum if (eltern, tag, dienst) in x ) @@ -393,8 +317,8 @@ class Elterndienstplaner: def _add_constraint_gesamtfairness( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable], - ziel_dienste: DefaultDict[str, DefaultDict[Dienst, float]], + x: Entscheidungsvariablen, + ziel_dienste: Zielverteilung, constraint_prefix: str ) -> Dict: """F3: Dienstübergreifende Fairness - verhindert Häufung bei einzelnen Eltern @@ -414,7 +338,7 @@ class Elterndienstplaner: """ fairness_abweichung_gesamt = {} - for eltern in self.eltern: + for eltern in self.daten.eltern: fairness_abweichung_gesamt[eltern] = pulp.LpVariable( f"fair_gesamt_{constraint_prefix}_{eltern.replace(' ', '_')}", lowBound=0) @@ -422,13 +346,13 @@ class Elterndienstplaner: # Tatsächliche Gesamtdienste für diesen Elternteil tatsaechliche_dienste_gesamt = pulp.lpSum( x[eltern, tag, dienst] - for tag in self.planungszeitraum - for dienst in self.dienste + for tag in self.daten.planungszeitraum + for dienst in self.daten.dienste if (eltern, tag, dienst) in x ) # Ziel-Gesamtdienste für diesen Elternteil (Summe über alle Dienste) - ziel_gesamt = sum(ziel_dienste[eltern][dienst] for dienst in self.dienste) + ziel_gesamt = sum(ziel_dienste[eltern][dienst] for dienst in self.daten.dienste) # Fairness-Constraints prob += (tatsaechliche_dienste_gesamt - ziel_gesamt <= @@ -442,7 +366,7 @@ class Elterndienstplaner: def _erstelle_zielfunktion( self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable], + x: Entscheidungsvariablen, fairness_abweichung_lokal: Dict, fairness_abweichung_global: Dict, fairness_abweichung_gesamt_global: Dict, @@ -460,8 +384,8 @@ class Elterndienstplaner: gewicht_f3_lokal = 0.25 * gewicht_lokal # Fairness-Terme zur Zielfunktion hinzufügen - for eltern in self.eltern: - for dienst in self.dienste: + for eltern in self.daten.eltern: + for dienst in self.daten.dienste: objective_terms.append(gewicht_f1 * fairness_abweichung_global[eltern, dienst]) objective_terms.append(gewicht_f2 * fairness_abweichung_lokal[eltern, dienst]) @@ -470,12 +394,12 @@ class Elterndienstplaner: objective_terms.append(gewicht_f3_lokal * fairness_abweichung_gesamt_lokal[eltern]) # P1: Bevorzugte Dienste (positiv belohnen) - for (eltern, tag, dienst), präf in self.präferenzen.items(): + for (eltern, tag, dienst), präf in self.daten.präferenzen.items(): if (eltern, tag, dienst) in x and präf == 1: # bevorzugt objective_terms.append(-5 * x[eltern, tag, dienst]) # P2: Abgelehnte Dienste (bestrafen) - for (eltern, tag, dienst), präf in self.präferenzen.items(): + for (eltern, tag, dienst), präf in self.daten.präferenzen.items(): if (eltern, tag, dienst) in x and präf == -1: # abgelehnt objective_terms.append(25 * x[eltern, tag, dienst]) @@ -491,9 +415,7 @@ class Elterndienstplaner: def erstelle_optimierungsmodell(self) -> Tuple[ pulp.LpProblem, - Dict[Tuple[str, date, Dienst], pulp.LpVariable], - DefaultDict[str, DefaultDict[Dienst, float]], - DefaultDict[str, DefaultDict[Dienst, float]] + Entscheidungsvariablen ]: """Erstellt das PuLP Optimierungsmodell @@ -504,9 +426,9 @@ class Elterndienstplaner: # Debugging: Verfügbarkeit prüfen print("\nDebug: Verfügbarkeit analysieren...") - for tag in self.planungszeitraum[:5]: # Erste 5 Tage - verfügbare = [e for e in self.eltern if self.verfügbarkeit.get((e, tag), True)] - benötigte = self.benoetigte_dienste.get(tag, []) + for tag in self.daten.planungszeitraum[:5]: # Erste 5 Tage + verfügbare = [e for e in self.daten.eltern if self.daten.verfügbarkeit.get((e, tag), True)] + benötigte = self.daten.benoetigte_dienste.get(tag, []) print(f" {tag}: Benötigt {len(benötigte)} Dienste {benötigte}, verfügbar: {verfügbare}") # LP Problem erstellen @@ -525,6 +447,9 @@ class Elterndienstplaner: ziel_dienste_global = self.berechne_faire_zielverteilung_global() ziel_dienste_lokal = self.berechne_faire_zielverteilung_lokal() + # Observer Pattern: Notify ausgabe about target distributions + self.ausgabe.setze_zielverteilungen(ziel_dienste_lokal, ziel_dienste_global) + # F2: Lokale Fairness-Constraints fairness_abweichung_lokal = self._add_fairness_constraints( prob, x, ziel_dienste_lokal, "lokal" @@ -550,10 +475,10 @@ class Elterndienstplaner: fairness_abweichung_gesamt_global, fairness_abweichung_gesamt_lokal) print(f"Modell erstellt mit {len(x)} Variablen und {len(prob.constraints)} Constraints") - return prob, x, ziel_dienste_lokal, ziel_dienste_global + return prob, x def löse_optimierung(self, prob: pulp.LpProblem, - x: Dict[Tuple[str, date, Dienst], pulp.LpVariable]) -> Optional[Dict[date, Dict[Dienst, List[str]]]]: + x: Entscheidungsvariablen) -> Optional[Dict[date, Dict[Dienst, List[Eltern]]]]: """Löst das Optimierungsproblem""" print("Löse Optimierungsproblem...") @@ -580,7 +505,7 @@ class Elterndienstplaner: return None # Lösung extrahieren - lösung: Dict[date, Dict[Dienst, List[str]]] = {} + lösung: Dict[date, Dict[Dienst, List[Eltern]]] = {} for (eltern, tag, dienst), var in x.items(): if var.varValue and var.varValue > 0.5: # Binary variable ist 1 if tag not in lösung: @@ -591,226 +516,6 @@ class Elterndienstplaner: return lösung - def schreibe_ausgabe_csv(self, datei: str, lösung: Dict[date, Dict[Dienst, List[str]]]) -> None: - """Schreibt die Lösung in die ausgabe.csv""" - AusgabeWriter.schreibe_ausgabe_csv(datei, lösung, self.planungszeitraum, self.dienste) - - def drucke_statistiken(self, lösung: Dict[date, Dict[Dienst, List[str]]]) -> None: - """Druckt Statistiken zur Lösung""" - print("\n" + "="*50) - print("STATISTIKEN") - print("="*50) - - # Dienste pro Eltern zählen - dienste_pro_eltern = defaultdict(lambda: defaultdict(int)) - for tag, tag_dienste in lösung.items(): - for dienst, eltern_liste in tag_dienste.items(): - for eltern in eltern_liste: - dienste_pro_eltern[eltern][dienst] += 1 - - # Gesamtübersicht - print("\nDienste pro Eltern:") - for eltern in sorted(self.eltern): - gesamt = sum(dienste_pro_eltern[eltern].values()) - dienste_detail = ', '.join(f"{dienst.kuerzel}:{dienste_pro_eltern[eltern][dienst]}" - for dienst in self.dienste if dienste_pro_eltern[eltern][dienst] > 0) - print(f" {eltern:15} {gesamt:3d} ({dienste_detail})") - - # Dienstfaktor-Analyse - print(f"\nDienstfaktoren im Planungszeitraum:") - for eltern in sorted(self.eltern): - faktor_summe = sum(self.dienstfaktoren[eltern][tag] for tag in self.planungszeitraum) - print(f" {eltern:15} {faktor_summe:.1f}") - - def visualisiere_praeferenz_verletzungen( - self, - lösung: Dict[date, Dict[Dienst, List[str]]] - ) -> None: - """Visualisiert verletzte Präferenzen als Tabelle - - Args: - lösung: Die tatsächliche Lösung nach Optimierung - """ - print("\n" + "="*110) - print("PRÄFERENZ-VERLETZUNGEN") - print("="*110) - - # Sammle alle zugeteilten Dienste pro Eltern - zugeteilte_dienste = defaultdict(lambda: defaultdict(list)) # eltern -> dienst -> [dates] - for tag, tag_dienste in lösung.items(): - for dienst, eltern_liste in tag_dienste.items(): - for eltern in eltern_liste: - zugeteilte_dienste[eltern][dienst].append(tag) - - # Sammle Präferenzen strukturiert - # praeferenzen_pro_eltern_dienst[eltern][dienst] = {datum: präf_wert} - praeferenzen_pro_eltern_dienst = defaultdict(lambda: defaultdict(dict)) - for (eltern, tag, dienst), präf in self.präferenzen.items(): - praeferenzen_pro_eltern_dienst[eltern][dienst][tag] = präf - - # Berechne Verletzungen - verletzungen = defaultdict(lambda: defaultdict(lambda: {'negativ': 0, 'positiv_nicht_erfuellt': 0})) - - for eltern in sorted(self.eltern): - for dienst in self.dienste: - zugeteilte_tage = zugeteilte_dienste[eltern][dienst] - praeferenzen_dienst = praeferenzen_pro_eltern_dienst[eltern][dienst] - - if not zugeteilte_tage: - continue # Keine Dienste zugeteilt - - # a) Negative Präferenzen die verletzt wurden - for tag in zugeteilte_tage: - if tag in praeferenzen_dienst and praeferenzen_dienst[tag] == -1: - verletzungen[eltern][dienst]['negativ'] += 1 - - # b) Positive Präferenzen nicht erfüllt (Dienst an nicht-präferiertem Tag) - # Sammle alle Tage mit positiver Präferenz für diesen Dienst - positive_praef_tage = {tag for tag, präf in praeferenzen_dienst.items() if präf == 1} - - if positive_praef_tage: # Es gibt positive Präferenzen - # Prüfe ob ALLE zugeteilten Dienste an nicht-präferierten Tagen sind - for tag in zugeteilte_tage: - if tag not in positive_praef_tage: - # Dienst wurde an nicht-präferiertem Tag zugeteilt - verletzungen[eltern][dienst]['positiv_nicht_erfuellt'] += 1 - - # Tabelle ausgeben - print(f"\n{'Eltern':<20} ", end='') - for dienst in self.dienste: - print(f"{dienst.kuerzel:>12}", end='') - print() - print(f"{'':20} ", end='') - for dienst in self.dienste: - print(f"{'neg, pos':>12}", end='') - print() - print("-" * (20 + 12 * len(self.dienste))) - - gesamt_negativ = defaultdict(int) - gesamt_positiv = defaultdict(int) - - for eltern in sorted(self.eltern): - print(f"{eltern:<20} ", end='') - for dienst in self.dienste: - neg = verletzungen[eltern][dienst]['negativ'] - pos = verletzungen[eltern][dienst]['positiv_nicht_erfuellt'] - - gesamt_negativ[dienst] += neg - gesamt_positiv[dienst] += pos - - # Farbcodierung - farbe = "" - reset = "" - if neg > 0 or pos > 0: - farbe = "\033[91m" if neg > 0 else "\033[93m" # Rot für negativ, Gelb für positiv - reset = "\033[0m" - - print(f"{farbe}{neg:>3}, {pos:>3}{reset:>6}", end='') - print() - - # Summenzeile - print("-" * (20 + 12 * len(self.dienste))) - print(f"{'SUMME':<20} ", end='') - for dienst in self.dienste: - neg = gesamt_negativ[dienst] - pos = gesamt_positiv[dienst] - - farbe = "" - reset = "" - if neg > 0 or pos > 0: - farbe = "\033[91m" if neg > 0 else "\033[93m" - reset = "\033[0m" - - print(f"{farbe}{neg:>3}, {pos:>3}{reset:>6}", end='') - print() - - print("\nLegende:") - print(" neg = Anzahl negativer Präferenzen (abgelehnte Tage), die verletzt wurden") - print(" pos = Anzahl Dienste an nicht-präferierten Tagen (obwohl präferierte Tage angegeben waren)") - print(" \033[91mRot\033[0m = Negative Präferenz verletzt") - print(" \033[93mGelb\033[0m = Positive Präferenz nicht erfüllt") - - def visualisiere_verteilungen( - self, - lösung: Dict[date, Dict[Dienst, List[str]]], - ziel_lokal: DefaultDict[str, DefaultDict[Dienst, float]], - ziel_global: DefaultDict[str, DefaultDict[Dienst, float]] - ) -> None: - """Visualisiert die Verteilungen als Tabelle zum Vergleich - - Args: - lösung: Die tatsächliche Lösung nach Optimierung - ziel_lokal: Lokale Zielverteilung (nur aktueller Planungszeitraum) - ziel_global: Globale Zielverteilung (inkl. Historie) - """ - # Tatsächliche Dienste zählen - tatsaechlich = defaultdict(lambda: defaultdict(int)) - for tag_dienste in lösung.values(): - for dienst, eltern_liste in tag_dienste.items(): - for eltern in eltern_liste: - tatsaechlich[eltern][dienst] += 1 - - print("\n" + "="*110) - print("VERTEILUNGSVERGLEICH: SOLL vs. IST") - print("="*110) - - for dienst in self.dienste: - print(f"\n{dienst.name} ({dienst.kuerzel}):") - print(f"{'Eltern':<20} {'Ziel Global':>12} {'Ziel Lokal':>12} {'Tatsächlich':>12} " - f"{'Δ Global':>12} {'Δ Lokal':>12}") - print("-" * 110) - - for eltern in sorted(self.eltern): - z_global = ziel_global[eltern][dienst] - z_lokal = ziel_lokal[eltern][dienst] - ist = tatsaechlich[eltern][dienst] - delta_global = ist - z_global - delta_lokal = ist - z_lokal - - # Farbcodierung für Abweichungen (ANSI-Codes) - farbe_global = "" - farbe_lokal = "" - reset = "" - - if abs(delta_global) > 0.5: - farbe_global = "\033[93m" if abs(delta_global) <= 1.0 else "\033[91m" # Gelb oder Rot - reset = "\033[0m" - - if abs(delta_lokal) > 0.5: - farbe_lokal = "\033[93m" if abs(delta_lokal) <= 1.0 else "\033[91m" - reset = "\033[0m" - - print(f"{eltern:<20} {z_global:>12.2f} {z_lokal:>12.2f} {ist:>12} " - f"{farbe_global}{delta_global:>+12.2f}{reset} {farbe_lokal}{delta_lokal:>+12.2f}{reset}") - - # Summen - summe_z_global = sum(ziel_global[e][dienst] for e in self.eltern) - summe_z_lokal = sum(ziel_lokal[e][dienst] for e in self.eltern) - summe_ist = sum(tatsaechlich[e][dienst] for e in self.eltern) - - print("-" * 110) - print(f"{'SUMME':<20} {summe_z_global:>12.2f} {summe_z_lokal:>12.2f} {summe_ist:>12} " - f"{summe_ist - summe_z_global:>+12.2f} {summe_ist - summe_z_lokal:>+12.2f}") - - # Gesamtstatistik - print("\n" + "="*110) - print("ZUSAMMENFASSUNG") - print("="*110) - - # Maximale Abweichungen finden - max_abw_global = 0 - max_abw_lokal = 0 - - for eltern in self.eltern: - for dienst in self.dienste: - ist = tatsaechlich[eltern][dienst] - max_abw_global = max(max_abw_global, abs(ist - ziel_global[eltern][dienst])) - max_abw_lokal = max(max_abw_lokal, abs(ist - ziel_lokal[eltern][dienst])) - - print(f"Maximale Abweichung von Global-Ziel: {max_abw_global:.2f} Dienste") - print(f"Maximale Abweichung von Lokal-Ziel: {max_abw_lokal:.2f} Dienste") - print("\nLegende: Δ = Tatsächlich - Ziel (positiv = mehr als Ziel, negativ = weniger als Ziel)") - def main() -> None: if len(sys.argv) < 4: @@ -826,28 +531,28 @@ def main() -> None: print("="*50) try: - planer = Elterndienstplaner() + # Create data model and load data + daten = ElterndienstplanerDaten() + daten.lade_daten(eingabe_datei, eltern_datei, vorherige_datei) - # Daten laden - planer.lade_eingabe_csv(eingabe_datei) - planer.lade_eltern_csv(eltern_datei) - if vorherige_datei: - planer.lade_vorherige_ausgaben_csv(vorherige_datei) + # Create output handler and optimization engine + ausgabe = ElterndienstAusgabe(daten) + planer = Elterndienstplaner(daten, ausgabe) # Optimierung - prob, x, ziel_lokal, ziel_global = planer.erstelle_optimierungsmodell() + prob, x = planer.erstelle_optimierungsmodell() lösung = planer.löse_optimierung(prob, x) if lösung is not None: # Ergebnisse ausgeben - planer.schreibe_ausgabe_csv(ausgabe_datei, lösung) - planer.drucke_statistiken(lösung) + ausgabe.schreibe_ausgabe_csv(ausgabe_datei, lösung) + ausgabe.drucke_statistiken(lösung) - # Visualisierung der Verteilungen - planer.visualisiere_verteilungen(lösung, ziel_lokal, ziel_global) + # Visualisierung der Verteilungen (uses Observer Pattern targets) + ausgabe.visualisiere_verteilungen(lösung) # Visualisierung der Präferenz-Verletzungen - planer.visualisiere_praeferenz_verletzungen(lösung) + ausgabe.visualisiere_praeferenz_verletzungen(lösung) print("\n✓ Planung erfolgreich abgeschlossen!") else: @@ -856,6 +561,8 @@ def main() -> None: except Exception as e: print(f"\n✗ Fehler: {e}") + import traceback + traceback.print_exc() sys.exit(1) diff --git a/typen.py b/typen.py new file mode 100644 index 0000000..ccf8500 --- /dev/null +++ b/typen.py @@ -0,0 +1,29 @@ +#!/usr/bin/env python3 +""" +Gemeinsame Type Aliases für den Elterndienstplaner + +Autor: Automatisch generiert +Datum: Dezember 2025 +""" + +from typing import Dict, Tuple, DefaultDict, TypeAlias, TYPE_CHECKING +from datetime import date +import pulp + +# Forward reference für zirkuläre Import-Vermeidung +if TYPE_CHECKING: + from datenmodell import Dienst + +# Definiert, welche Namen bei "from typen import *" exportiert werden +__all__ = ['Eltern', 'Zielverteilung', 'Entscheidungsvariablen'] + +# Type Alias für Elternnamen +Eltern: TypeAlias = str + +# Type Alias für Zielverteilungen +# Struktur: DefaultDict[Elternname, DefaultDict[Dienst, Anzahl]] +Zielverteilung: TypeAlias = DefaultDict[Eltern, DefaultDict['Dienst', float]] + +# Type Alias für Entscheidungsvariablen des Optimierungsproblems +# Struktur: Dict[(Eltern, Datum, Dienst), LP-Variable] +Entscheidungsvariablen: TypeAlias = Dict[Tuple[Eltern, date, 'Dienst'], pulp.LpVariable]