commit a37a680b2083d4a7efc2d5f4dde9c67193357c03 Author: Jan Hoheisel Date: Sun Dec 21 21:57:23 2025 +0100 init diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..4b8fac4 --- /dev/null +++ b/README.md @@ -0,0 +1,163 @@ +# Elterndienstplaner + +Automatische Zuteilung von Elterndiensten im Kinderladen unter Berücksichtigung von Fairness und Präferenzen. + +## Verwendung + +```bash +./elterndienstplaner.py [] +``` + +**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 + +## Dienste + +- **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) + +Die Planung erfolgt für einen Kalendermonat. + +## Eingabedateien + +### eingabe.csv + +Informationen zu benötigten Diensten und Verfügbarkeit/Präferenzen der 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 + +### eltern.csv + +Dienstfaktoren (= Anzahl betreuter Kinder) pro Elternteil und Zeitraum. + +**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 +``` + +**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 (müssen erfüllt sein) + +- **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 + +### 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) + +- **F2 (Lokal)**: Dienste proportional zum Dienstfaktor im **aktuellen Monat** + - Nur aktueller Planungszeitraum + - Gewichtung: 60% (zu Jahresbeginn) → 40% (zu Jahresende) + +- **F3 (Dienstübergreifend)**: Gleiche Gesamtanzahl über alle Diensttypen + - Verhindert Häufung bei einzelnen Eltern + +**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 new file mode 100644 index 0000000..e69de29 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/csv_io.py b/csv_io.py new file mode 100644 index 0000000..8052e6b --- /dev/null +++ b/csv_io.py @@ -0,0 +1,281 @@ +#!/usr/bin/env python3 +""" +CSV I/O Module für Elterndienstplaner +Trennt CSV-Parsing und -Schreiben von der Business-Logik +""" + +import csv +from datetime import datetime, date, timedelta +from typing import Dict, List, Tuple, DefaultDict +from collections import defaultdict + + +class EingabeParser: + """Parst CSV-Eingabedateien für den Elterndienstplaner""" + + @staticmethod + def parse_eingabe_csv(datei: str, dienste_lookup) -> Tuple[ + List[str], # eltern + List[date], # tage + Dict[date, List], # benoetigte_dienste (Dienst-Objekte) + Dict[Tuple[str, date], bool], # verfügbarkeit + Dict[Tuple[str, date, any], int] # präferenzen (Dienst-Objekt als Key) + ]: + """ + Lädt die eingabe.csv mit Terminen und Präferenzen + + Args: + datei: Pfad zur eingabe.csv + dienste_lookup: Funktion (str) -> Dienst zum Auflösen von Kürzeln + + Returns: + Tuple mit (eltern, tage, benoetigte_dienste, verfügbarkeit, präferenzen) + """ + print(f"Lade Eingabedaten aus {datei}...") + + eltern = [] + tage = [] + benoetigte_dienste = {} + verfügbarkeit = {} + präferenzen = {} + + with open(datei, 'r', encoding='utf-8') as f: + reader = csv.reader(f) + header = next(reader) + + # Eltern aus Header extrahieren (ab Spalte 3) + eltern = [name.strip() for name in header[3:] if name.strip()] + print(f"Gefundene Eltern: {eltern}") + + for row in reader: + if len(row) < 3: + continue + + datum = row[0].strip() + wochentag = row[1].strip() + dienste_str = row[2].strip() + + # Datum parsen + try: + datum_obj = datetime.strptime(datum, '%Y-%m-%d').date() + tage.append(datum_obj) + except ValueError: + print(f"Warnung: Ungültiges Datum {datum}") + continue + + # Benötigte Dienste + benoetigte_dienste[datum_obj] = [ + dienste_lookup(kuerzel) for kuerzel in dienste_str + ] + + # Verfügbarkeit und Präferenzen der Eltern + for i, eltern_name in enumerate(eltern): + if i + 3 < len(row): + präf_str = row[i + 3].strip() + + # Verfügbarkeit prüfen + if präf_str == 'x': + verfügbarkeit[(eltern_name, datum_obj)] = False + else: + verfügbarkeit[(eltern_name, datum_obj)] = True + + # Präferenzen parsen + _parse_präferenzen_string( + eltern_name, datum_obj, präf_str, + präferenzen, dienste_lookup + ) + else: + # Standard: verfügbar, keine Präferenzen + verfügbarkeit[(eltern_name, datum_obj)] = True + + tage.sort() + print(f"Zeitraum: {tage[0]} bis {tage[-1]} ({len(tage)} Tage)") + + return eltern, tage, benoetigte_dienste, verfügbarkeit, präferenzen + + @staticmethod + def parse_eltern_csv(datei: str) -> Dict[str, DefaultDict[date, float]]: + """ + Lädt die eltern.csv mit Dienstfaktoren + + Args: + datei: Pfad zur eltern.csv + + Returns: + Dictionary mit Dienstfaktoren für alle Tage in den definierten Zeiträumen + (DefaultDict gibt 0 für Tage außerhalb der Zeiträume zurück) + """ + print(f"Lade Elterndaten aus {datei}...") + + dienstfaktoren = {} + + with open(datei, 'r', encoding='utf-8') as f: + reader = csv.reader(f) + header = next(reader) + + for row in reader: + if len(row) < 4: + continue + + eltern_name = row[0].strip() + + # Initialisiere mit DefaultDict (Standard: 0) + dienstfaktoren[eltern_name] = defaultdict(float) + + # Alle Zeiträume einlesen (jeweils 3 Spalten: Beginn, Ende, Faktor) + for i in range(1, len(row), 3): + if i + 2 < len(row) and row[i].strip() and row[i + 1].strip(): + try: + beginn = datetime.strptime(row[i].strip(), '%Y-%m-%d').date() + ende = datetime.strptime(row[i + 1].strip(), '%Y-%m-%d').date() + faktor = float(row[i + 2].strip()) if row[i + 2].strip() else 0 + + # Faktor für alle Tage im Zeitraum setzen/überschreiben + aktueller_tag = beginn + while aktueller_tag <= ende: + dienstfaktoren[eltern_name][aktueller_tag] = faktor + aktueller_tag += timedelta(days=1) + + except (ValueError, IndexError): + continue + + print(f"Dienstfaktoren geladen für {len(dienstfaktoren)} Eltern") + + return dienstfaktoren + + @staticmethod + def parse_vorherige_ausgaben_csv( + datei: str, + eltern: List[str], + dienste: List, # List[Dienst] + ) -> List[Tuple[date, str, any]]: # historische_dienste (mit Dienst-Objekt) + """ + Lädt vorherige-ausgaben.csv für Fairness-Constraints + + Args: + datei: Pfad zur vorherige-ausgaben.csv + eltern: Liste der Elternnamen + dienste: Liste der Dienst-Objekte + + Returns: + Liste der historischen Dienste + """ + print(f"Lade vorherige Ausgaben aus {datei}...") + + historische_dienste = [] + + try: + with open(datei, 'r', encoding='utf-8') as f: + reader = csv.reader(f) + header = next(reader) + + # Dienst-Spalten finden + dienst_spalten = {} + for i, col_name in enumerate(header[2:], 2): # Ab Spalte 2 (nach Datum, Wochentag) + for dienst in dienste: + if dienst.name.lower() in col_name.lower() or dienst.kuerzel == col_name: + dienst_spalten[dienst] = i + break + + for row in reader: + if len(row) < 3: + continue + + # Datum parsen + try: + datum = datetime.strptime(row[0].strip(), '%Y-%m-%d').date() + except ValueError: + continue + + # Zugeteilte Dienste mit Datum speichern + for dienst, spalte_idx in dienst_spalten.items(): + if spalte_idx < len(row) and row[spalte_idx].strip(): + # Mehrere Eltern können in einer Zelle stehen (durch " und " getrennt) + eltern_liste = row[spalte_idx].strip().split(' und ') + for eltern_name in eltern_liste: + if eltern_name in eltern: + # Historische Dienste mit Datum speichern + historische_dienste.append((datum, eltern_name, dienst)) + + except FileNotFoundError: + print("Keine vorherigen Ausgaben gefunden - starte ohne historische Daten") + + print(f"Historische Dienste mit Datum: {len(historische_dienste)} Einträge") + + return historische_dienste + + +class AusgabeWriter: + """Schreibt Optimierungsergebnisse in CSV-Dateien""" + + @staticmethod + def schreibe_ausgabe_csv( + datei: str, + lösung: Dict[date, Dict[any, List[str]]], # Dienst-Objekt als Key + tage: List[date], + dienste: List # List[Dienst] + ) -> None: + """ + Schreibt die Lösung in die ausgabe.csv + + Args: + datei: Pfad zur ausgabe.csv + lösung: Dictionary mit Zuteilungen {datum: {dienst: [eltern]}} + tage: Liste aller Planungstage + dienste: Liste der Dienst-Objekte + """ + print(f"Schreibe Ergebnisse nach {datei}...") + + with open(datei, 'w', newline='', encoding='utf-8') as f: + writer = csv.writer(f) + + # Header schreiben + header = ['Datum', 'Wochentag'] + [dienst.name for dienst in dienste] + writer.writerow(header) + + # Daten schreiben + for tag in sorted(tage): + wochentag = ['Montag', 'Dienstag', 'Mittwoch', 'Donnerstag', + 'Freitag', 'Samstag', 'Sonntag'][tag.weekday()] + + row = [tag.strftime('%Y-%m-%d'), wochentag] + + for dienst in dienste: + if tag in lösung and dienst in lösung[tag]: + eltern_str = ' und '.join(lösung[tag][dienst]) + else: + eltern_str = '' + row.append(eltern_str) + + writer.writerow(row) + + print("Ausgabe erfolgreich geschrieben!") + + +# Hilfsfunktionen + +def _parse_präferenzen_string( + eltern: str, + datum: date, + präf_str: str, + präferenzen: Dict, + dienste_lookup +) -> None: + """Parst Präferenzstring wie 'F+P-E+' und fügt zu präferenzen hinzu""" + i = 0 + while i < len(präf_str): + if i + 1 < len(präf_str): + dienst = dienste_lookup(präf_str[i]) + if dienst and i + 1 < len(präf_str): + if präf_str[i + 1] == '+': + präferenzen[(eltern, datum, dienst)] = 1 + i += 2 + elif präf_str[i + 1] == '-': + präferenzen[(eltern, datum, dienst)] = -1 + i += 2 + else: + i += 1 + else: + i += 1 + else: + i += 1 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 new file mode 100755 index 0000000..6429b90 --- /dev/null +++ b/elterndienstplaner.py @@ -0,0 +1,570 @@ +#!/usr/bin/env python3 +""" +Elterndienstplaner - Optimale Zuteilung von Elterndiensten + +Autor: Automatisch generiert +Datum: Dezember 2025 +""" + +import sys +import pulp +from datetime import timedelta, date +from collections import defaultdict +from typing import Dict, List, Tuple, DefaultDict, Optional + +from datenmodell import ElterndienstplanerDaten, Dienst, Eltern, Zielverteilung, Entscheidungsvariablen +from ausgabe import ElterndienstAusgabe + + +class Elterndienstplaner: + """Optimierungs-Engine für Elterndienstplanung""" + + def __init__(self, daten: ElterndienstplanerDaten, ausgabe: ElterndienstAusgabe) -> None: + self.daten = daten + self.ausgabe = ausgabe + + def berechne_faire_zielverteilung_global(self) -> Zielverteilung: + """Berechnet die faire Zielanzahl von Diensten für den Planungszeitraum + basierend auf globaler Fairness (Historie + aktueller Planungszeitraum). + + Gibt die Ziel-Dienstanzahl für den aktuellen Planungszeitraum zurück, + korrigiert um bereits geleistete Dienste. Kann negativ sein, wenn bereits + mehr Dienste geleistet wurden als fair wäre.""" + + 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.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.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.daten.historische_dienste + if d == dienst + ] + + print(f" Gefundene historische {dienst.kuerzel}-Dienste: {len(historische_dienste_dieses_typs)}") + + # Gruppiere nach Datum + dienste_pro_tag = defaultdict(list) + for datum, eltern in historische_dienste_dieses_typs: + dienste_pro_tag[datum].append(eltern) + + # Für jeden historischen Tag faire Umverteilung berechnen + for tag, geleistete_eltern in dienste_pro_tag.items(): + anzahl_dienste = len(geleistete_eltern) # Anzahl Dienste an diesem Tag + + # Dienstfaktoren aller Eltern für diesen historischen Tag berechnen + gesamt_dienstfaktor_tag = 0 + + 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.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.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.daten.planungszeitraum: + # Prüfe ob an diesem Tag der Dienst benötigt wird + if dienst not in self.daten.benoetigte_dienste.get(tag, []): + continue + + benoetigte_dienste_planungszeitraum += dienst.personen_anzahl + + # Dienstfaktoren aller Eltern für diesen Tag berechnen + dienstfaktoren = {} + gesamt_dienstfaktor_tag = 0 + + 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.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.daten.eltern: + # Berechne vorherige Dienste on-the-fly aus historischen Diensten + vorherige_anzahl = sum( + 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) -> 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: 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.daten.dienstfaktoren[e][tag] for tag in self.daten.planungszeitraum) + for e in self.daten.eltern + ) + + if summe_dienstfaktor_planungszeitraum_alle_eltern == 0: + print(" WARNUNG: Gesamtdienstfaktor ist 0, keine lokale Zielverteilung möglich") + return ziel_dienste_lokal + + # Für jeden Dienst die lokale faire Verteilung berechnen + for dienst in self.daten.dienste: + # Anzahl benötigter Dienste im aktuellen Planungszeitraum + benoetigte_dienste_planungszeitraum = sum( + 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 + + if benoetigte_dienste_planungszeitraum > 0: + print(f" {dienst.kuerzel}: {benoetigte_dienste_planungszeitraum} Dienste benötigt") + + for eltern in self.daten.eltern: + # Dienstfaktor für diesen Elternteil im aktuellen Planungszeitraum + summe_dienstfaktor_planungszeitraum = sum( + self.daten.dienstfaktoren[eltern][tag] for tag in self.daten.planungszeitraum + ) + + if summe_dienstfaktor_planungszeitraum > 0: + anteil = summe_dienstfaktor_planungszeitraum / summe_dienstfaktor_planungszeitraum_alle_eltern + faire_zuteilung = anteil * benoetigte_dienste_planungszeitraum + ziel_dienste_lokal[eltern][dienst] = faire_zuteilung + + return ziel_dienste_lokal + + def _erstelle_entscheidungsvariablen(self) -> Entscheidungsvariablen: + """Erstellt die binären Entscheidungsvariablen x[eltern, tag, dienst]""" + 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' + ) + return x + + def _add_constraint_ein_dienst_pro_woche( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen + ) -> None: + """C1: Je Eltern und Dienst nur einmal die Woche (Woche = Montag bis Sonntag)""" + 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.daten.planungszeitraum[-1] + + while woche_start <= letzter_tag: + woche_ende = woche_start + timedelta(days=6) # Sonntag + + 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.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.daten.planungszeitraum: + if woche_start <= tag <= woche_ende: + if (eltern, tag, dienst) in x: + woche_vars.append(x[eltern, tag, dienst]) + + # Constraint: Historische + geplante Dienste <= 1 + if woche_vars: + prob += pulp.lpSum(woche_vars) <= 1 - historische_dienste_in_woche, \ + f"C1_{eltern.replace(' ', '_')}_{dienst.kuerzel}_w{woche_nr}" + + woche_start += timedelta(days=7) + woche_nr += 1 + + def _add_constraint_ein_dienst_pro_tag( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen + ) -> None: + """C2: Je Eltern nur einen Dienst am Tag""" + for eltern in self.daten.eltern: + for tag in self.daten.planungszeitraum: + tag_vars = [] + for dienst in self.daten.dienste: + if (eltern, tag, dienst) in x: + tag_vars.append(x[eltern, tag, dienst]) + + if tag_vars: + prob += pulp.lpSum(tag_vars) <= 1, f"C2_{eltern.replace(' ', '_')}_{tag}" + + def _add_constraint_verfuegbarkeit( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen + ) -> None: + """C3: Dienste nur verfügbaren Eltern zuteilen""" + 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}" + + def _add_constraint_dienst_bedarf( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen + ) -> None: + """C4: Alle benötigten Dienste müssen zugeteilt werden""" + for tag in self.daten.planungszeitraum: + for dienst in self.daten.benoetigte_dienste.get(tag, []): + dienst_vars = [] + for eltern in self.daten.eltern: + if (eltern, tag, dienst) in x: + # Prüfe ob Eltern verfügbar + if self.daten.verfügbarkeit.get((eltern, tag), True): + dienst_vars.append(x[eltern, tag, dienst]) + + if dienst_vars: + # Anzahl benötigter Personen pro Dienst (aus Dienst-Objekt) + benoetigte_personen = dienst.personen_anzahl + prob += pulp.lpSum(dienst_vars) == benoetigte_personen, \ + f"Bedarf_{tag}_{dienst.kuerzel}" + + def _add_fairness_constraints( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen, + ziel_dienste: Zielverteilung, + constraint_prefix: str + ) -> Dict: + """Erstellt Fairness-Variablen und fügt Fairness-Constraints hinzu + + Args: + prob: Das LP-Problem + x: Die Entscheidungsvariablen + ziel_dienste: Die Zielverteilung der Dienste + constraint_prefix: Präfix für Constraint-Namen ('lokal' oder 'global') + + Returns: + Dictionary mit Fairness-Abweichungsvariablen + """ + # Hilfsvariablen für Fairness-Abweichungen erstellen + fairness_abweichung = {} + + 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.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.daten.planungszeitraum + if (eltern, tag, dienst) in x + ) + + # Ziel für diese Fairness-Variante + ziel = ziel_dienste[eltern][dienst] + prob += (zugeteilte_dienste_planungszeitraum - ziel <= + fairness_abweichung[eltern, dienst]) + prob += (ziel - zugeteilte_dienste_planungszeitraum <= + fairness_abweichung[eltern, dienst]) + + return fairness_abweichung + + def _add_constraint_gesamtfairness( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen, + ziel_dienste: Zielverteilung, + constraint_prefix: str + ) -> Dict: + """F3: Dienstübergreifende Fairness - verhindert Häufung bei einzelnen Eltern + + Berechnet die Abweichung der Gesamtdienstanzahl (über alle Diensttypen) + vom fairen Gesamtziel. Dies verhindert, dass einzelne Eltern über alle + Diensttypen hinweg überproportional viele Dienste bekommen. + + Args: + prob: Das LP-Problem + x: Die Entscheidungsvariablen + ziel_dienste: Die Zielverteilung (global oder lokal) + constraint_prefix: Präfix für Constraint-Namen ('lokal' oder 'global') + + Returns: + Dictionary mit Gesamt-Fairness-Abweichungsvariablen + """ + fairness_abweichung_gesamt = {} + + for eltern in self.daten.eltern: + fairness_abweichung_gesamt[eltern] = pulp.LpVariable( + f"fair_gesamt_{constraint_prefix}_{eltern.replace(' ', '_')}", + lowBound=0) + + # Tatsächliche Gesamtdienste für diesen Elternteil + tatsaechliche_dienste_gesamt = pulp.lpSum( + x[eltern, tag, dienst] + 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.daten.dienste) + + # Fairness-Constraints + prob += (tatsaechliche_dienste_gesamt - ziel_gesamt <= + fairness_abweichung_gesamt[eltern]) + prob += (ziel_gesamt - tatsaechliche_dienste_gesamt <= + fairness_abweichung_gesamt[eltern]) + + return fairness_abweichung_gesamt + + + def _erstelle_zielfunktion( + self, + prob: pulp.LpProblem, + x: Entscheidungsvariablen, + fairness_abweichung_lokal: Dict, + fairness_abweichung_global: Dict, + fairness_abweichung_gesamt_global: Dict, + fairness_abweichung_gesamt_lokal: Dict + ) -> None: + """Erstellt die Zielfunktion mit Fairness und Präferenzen""" + objective_terms = [] + + # Fairness-Gewichtung + gewicht_global = 40 + gewicht_lokal = 60 + gewicht_f1 = gewicht_global + gewicht_f2 = gewicht_lokal + gewicht_f3_global = 0.25 * gewicht_global + gewicht_f3_lokal = 0.25 * gewicht_lokal + + # Fairness-Terme zur Zielfunktion hinzufügen + 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]) + + # F3: Gesamtfairness (dienstübergreifend) - global und lokal + objective_terms.append(gewicht_f3_global * fairness_abweichung_gesamt_global[eltern]) + objective_terms.append(gewicht_f3_lokal * fairness_abweichung_gesamt_lokal[eltern]) + + # P1: Bevorzugte Dienste (positiv belohnen) + 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.daten.präferenzen.items(): + if (eltern, tag, dienst) in x and präf == -1: # abgelehnt + objective_terms.append(25 * x[eltern, tag, dienst]) + + # Zielfunktion setzen + if objective_terms: + prob += pulp.lpSum(objective_terms) + else: + # Fallback: Minimiere Gesamtanzahl Dienste + prob += pulp.lpSum([var for var in x.values()]) + + print(f"Verwende Gewichtung: F1 (global) = {gewicht_f1}, F2 (lokal) = {gewicht_f2}, " + f"F3_global = {gewicht_f3_global}, F3_lokal = {gewicht_f3_lokal}") + + def erstelle_optimierungsmodell(self) -> Tuple[ + pulp.LpProblem, + Entscheidungsvariablen + ]: + """Erstellt das PuLP Optimierungsmodell + + Returns: + Tuple mit (prob, x, ziel_dienste_lokal, ziel_dienste_global) + """ + print("Erstelle Optimierungsmodell...") + + # Debugging: Verfügbarkeit prüfen + print("\nDebug: Verfügbarkeit analysieren...") + 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 + prob = pulp.LpProblem("Elterndienstplaner", pulp.LpMinimize) + + # Entscheidungsvariablen erstellen + x = self._erstelle_entscheidungsvariablen() + + # Grundlegende Constraints hinzufügen + self._add_constraint_ein_dienst_pro_woche(prob, x) + self._add_constraint_ein_dienst_pro_tag(prob, x) + self._add_constraint_verfuegbarkeit(prob, x) + self._add_constraint_dienst_bedarf(prob, x) + + # Fairness-Constraints + 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" + ) + + # F1: Globale Fairness-Constraints + fairness_abweichung_global = self._add_fairness_constraints( + prob, x, ziel_dienste_global, "global" + ) + + # F3: Dienstübergreifende Fairness - Global + fairness_abweichung_gesamt_global = self._add_constraint_gesamtfairness( + prob, x, ziel_dienste_global, "global" + ) + + # F3: Dienstübergreifende Fairness - Lokal + fairness_abweichung_gesamt_lokal = self._add_constraint_gesamtfairness( + prob, x, ziel_dienste_lokal, "lokal" + ) + + # Zielfunktion erstellen + self._erstelle_zielfunktion(prob, x, fairness_abweichung_lokal, fairness_abweichung_global, + fairness_abweichung_gesamt_global, fairness_abweichung_gesamt_lokal) + + print(f"Modell erstellt mit {len(x)} Variablen und {len(prob.constraints)} Constraints") + return prob, x + + def löse_optimierung(self, prob: pulp.LpProblem, + x: Entscheidungsvariablen) -> Optional[Dict[date, Dict[Dienst, List[Eltern]]]]: + """Löst das Optimierungsproblem""" + print("Löse Optimierungsproblem...") + + # Solver wählen (verfügbare Solver testen) + solver = None + try: + print("Versuche CBC Solver...") + solver = pulp.PULP_CBC_CMD(msg=0, timeLimit=10) # Standard CBC Solver + except: + try: + print("Versuche GLPK Solver...") + solver = pulp.GLPK_CMD(msg=0) # GLPK falls verfügbar + except: + print("Kein spezifizierter Solver verfügbar, verwende Standard.") + solver = None # Default Solver + + prob.solve(solver) + + status = pulp.LpStatus[prob.status] + print(f"Optimierung abgeschlossen: {status}") + + if prob.status != pulp.LpStatusOptimal: + print("WARNUNG: Keine optimale Lösung gefunden!") + return None + + # Lösung extrahieren + 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: + lösung[tag] = {} + if dienst not in lösung[tag]: + lösung[tag][dienst] = [] + lösung[tag][dienst].append(eltern) + + return lösung + + +def main() -> None: + if len(sys.argv) < 4: + print("Usage: ./elterndienstplaner.py []") + sys.exit(1) + + eingabe_datei = sys.argv[1] + eltern_datei = sys.argv[2] + ausgabe_datei = sys.argv[3] + vorherige_datei = sys.argv[4] if len(sys.argv) > 4 else None + + print("Elterndienstplaner gestartet") + print("="*50) + + try: + # Create data model and load data + daten = ElterndienstplanerDaten() + daten.lade_daten(eingabe_datei, eltern_datei, vorherige_datei) + + # Create output handler and optimization engine + ausgabe = ElterndienstAusgabe(daten) + planer = Elterndienstplaner(daten, ausgabe) + + # Optimierung + prob, x = planer.erstelle_optimierungsmodell() + lösung = planer.löse_optimierung(prob, x) + + if lösung is not None: + # Ergebnisse ausgeben + ausgabe.schreibe_ausgabe_csv(ausgabe_datei, lösung) + ausgabe.drucke_statistiken(lösung) + + # Visualisierung der Verteilungen (uses Observer Pattern targets) + ausgabe.visualisiere_verteilungen(lösung) + + # Visualisierung der Präferenz-Verletzungen + ausgabe.visualisiere_praeferenz_verletzungen(lösung) + + print("\n✓ Planung erfolgreich abgeschlossen!") + else: + print("\n✗ Fehler: Keine gültige Lösung gefunden!") + sys.exit(1) + + except Exception as e: + print(f"\n✗ Fehler: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/elterndienstplaner.sh b/elterndienstplaner.sh new file mode 100755 index 0000000..7b36b36 --- /dev/null +++ b/elterndienstplaner.sh @@ -0,0 +1,2 @@ +DIR=/home/jwit/privat/elterndienstplaner/ +$DIR/.venv/bin/python $DIR/elterndienstplaner.py eingabe.csv eltern.csv ausgabe.csv vorherige-ausgaben.csv diff --git a/test/simple/ausgabe.csv b/test/simple/ausgabe.csv new file mode 100644 index 0000000..2bbbedc --- /dev/null +++ b/test/simple/ausgabe.csv @@ -0,0 +1,24 @@ +Datum,Wochentag,Frühstücksdienst,Putznotdienst,Essensausgabenotdienst,Kochen,Elternabend +2026-01-01,Donnerstag,Ben & Nele,Marie,Jonas,, +2026-01-02,Freitag,Jonas,Ben & Nele,Laura,, +2026-01-03,Samstag,Marie,Jonas,Ben & Nele,, +2026-01-06,Dienstag,Laura,Jonas,Marie,, +2026-01-07,Mittwoch,Paula,Ben & Nele,Jonas,, +2026-01-08,Donnerstag,Jonas,Paula,Erwin,Ben & Nele, +2026-01-09,Freitag,Ben & Nele,Marie,Paula,, +2026-01-10,Samstag,Marie,Laura,Ben & Nele,, +2026-01-13,Dienstag,Ben & Nele,Jonas,Laura,, +2026-01-14,Mittwoch,Paula,Erwin,Jonas,, +2026-01-15,Donnerstag,Laura,Paula,Marie,, +2026-01-16,Freitag,Marie,Laura,Ben & Nele,, +2026-01-17,Samstag,Erwin,Ben & Nele,Paula,, +2026-01-20,Dienstag,Jonas,Erwin,Marie,, +2026-01-21,Mittwoch,Erwin,Ben & Nele,Jonas,, +2026-01-22,Donnerstag,Ben & Nele,Marie,Laura,Jonas, +2026-01-23,Freitag,Marie,Laura,Ben & Nele,, +2026-01-24,Samstag,Laura,Jonas,Erwin,,Ben & Nele Marie +2026-01-27,Dienstag,Erwin,Ben & Nele,Jonas,, +2026-01-28,Mittwoch,Jonas,Erwin,Marie,, +2026-01-29,Donnerstag,Marie,Paula,Ben & Nele,, +2026-01-30,Freitag,Ben & Nele,Jonas,Paula,, +2026-01-31,Samstag,Paula,Marie,Erwin,, diff --git a/test/simple/eingabe.csv b/test/simple/eingabe.csv new file mode 100644 index 0000000..36f3dae --- /dev/null +++ b/test/simple/eingabe.csv @@ -0,0 +1,24 @@ +Datum , Wochentag , Dienste, Erwin, Paula, Laura, Ben & Nele, Jonas, Marie +2026-01-01, Mittwoch , FPE , x , F+ , P- , , E+ , P+ +2026-01-02, Donnerstag, FPE , x , , E+ , P+ , F+ , P- +2026-01-03, Freitag , FPE , x , F+ , , F-E+ , P+ , E- +2026-01-06, Montag , FPE , F- , P+ , F+ , E+ , , F- +2026-01-07, Dienstag , FPE , P+ , F+ , P- , F+P+ , E+ , F- +2026-01-08, Mittwoch , FPEK , E+ , P+ , F+ , K+ , P- , E+ +2026-01-09, Donnerstag, FPE , F+ , , E- , P+F+ , , P+ +2026-01-10, Freitag , FPE , , F+ , P+ , E+ , F- , +2026-01-13, Montag , FPE , F+ , P- , E+ , F+ , P+ , E- +2026-01-14, Dienstag , FPE , P+ , F+ , F- , P+E+ , E+ , P- +2026-01-15, Mittwoch , FPE , E+ , P+ , F+ , F- , , E+ +2026-01-16, Donnerstag, FPE , F- , , P+ , E+F+ , P- , F+ +2026-01-17, Freitag , FPE , F+ , F+ , E+ , P+ , , F- +2026-01-20, Montag , FPE , P+ , x , F- , F+E+ , F+ , P+ +2026-01-21, Dienstag , FPE , F+ , x , P+ , P+ , E+ , F- +2026-01-22, Mittwoch , FPEK , , x , E+ , K+F+ , P+ , K- +2026-01-23, Donnerstag, FPE , F+ , x , F- , E+ , P+ , F+ +2026-01-24, Freitag , FPEA , P+ , x , F+ , A+F+ , E- , A+ +2026-01-27, Montag , FPE , F+ , P- , x , P+F+ , E+ , P- +2026-01-28, Dienstag , FPE , E+ , F+ , x , F- , P+ , E+ +2026-01-29, Mittwoch , FPE , F- , P+ , x , E+P+ , , F+ +2026-01-30, Donnerstag, FPE , F+ , , x , F+ , P+ , E- +2026-01-31, Freitag , FPE , P+ , F+ , x , E+ , F- , P+ diff --git a/test/simple/eltern.csv b/test/simple/eltern.csv new file mode 100644 index 0000000..c301023 --- /dev/null +++ b/test/simple/eltern.csv @@ -0,0 +1,7 @@ +Name_Kind(er),Zeitraum_Beginn,Zeitraum_Ende,Dienstfaktor,Zeitraum_Beginn2,Zeitraum_Ende2,Dienstfaktor2,Zeitraum_Beginn3,Zeitraum_Ende3,Dienstfaktor3 +Erwin,2025-09-01,2026-07-31,1,,,,, +Paula,2025-09-01,2026-07-31,1,,,,, +Laura,2025-09-01,2026-07-31,1,,,,, +Ben & Nele,2025-09-01,2026-07-31,2,,,,, +Jonas,2025-09-01,2026-07-31,1,,,,, +Marie,2025-09-01,2026-07-31,1,,,,, diff --git a/test/simple/vorherige-ausgaben.csv b/test/simple/vorherige-ausgaben.csv new file mode 100644 index 0000000..9cf2bbe --- /dev/null +++ b/test/simple/vorherige-ausgaben.csv @@ -0,0 +1,37 @@ +Datum,Wochentag,Frühstücksdienst,Putznotdienst,Essensausgabenotdienst,Kochen,Elternabend +2025-12-02,Montag,Paula,Erwin,Laura,, +2025-12-03,Dienstag,Laura,Ben & Nele,Paula,, +2025-12-04,Mittwoch,Erwin,Paula,Ben & Nele,, +2025-12-05,Donnerstag,Ben & Nele,Laura,Erwin,, +2025-12-06,Freitag,Jonas,Marie,Laura,, +2025-12-09,Montag,Laura,Ben & Nele,Jonas,, +2025-12-10,Dienstag,Marie,Paula,Ben & Nele,, +2025-12-11,Mittwoch,Ben & Nele,Laura,Erwin,Paula, +2025-12-12,Donnerstag,Paula,Jonas,Marie,, +2025-12-13,Freitag,Laura,Jonas,Paula,, +2025-12-16,Montag,Erwin,Marie,Ben & Nele,, +2025-12-17,Dienstag,Ben & Nele,Laura,Jonas,, +2025-12-18,Mittwoch,Marie,Erwin,Laura,, +2025-12-19,Donnerstag,Laura,Ben & Nele,Paula,, +2025-12-20,Freitag,Jonas,Paula,Ben & Nele,Ben & Nele,Laura Erwin +2025-11-01,Freitag,Laura,Erwin,Ben & Nele,, +2025-11-04,Montag,Paula,Ben & Nele,Laura,, +2025-11-05,Dienstag,Erwin,Paula,Ben & Nele,, +2025-11-06,Mittwoch,Ben & Nele,Laura,Erwin,Erwin, +2025-11-07,Donnerstag,Paula,Erwin,Laura,, +2025-11-08,Freitag,Laura,Ben & Nele,Paula,, +2025-11-11,Montag,Erwin,Paula,Ben & Nele,, +2025-11-12,Dienstag,Ben & Nele,Laura,Erwin,, +2025-11-13,Mittwoch,Paula,Erwin,Laura,, +2025-11-14,Donnerstag,Laura,Ben & Nele,Paula,, +2025-11-15,Freitag,Erwin,Paula,Ben & Nele,, +2025-11-18,Montag,Ben & Nele,Laura,Erwin,, +2025-11-19,Dienstag,Paula,Erwin,Laura,, +2025-11-20,Mittwoch,Laura,Ben & Nele,Paula,Paula, +2025-11-21,Donnerstag,Erwin,Paula,Ben & Nele,, +2025-11-22,Freitag,Ben & Nele,Laura,Erwin,, +2025-11-25,Montag,Paula,Erwin,Laura,, +2025-11-26,Dienstag,Laura,Ben & Nele,Paula,, +2025-11-27,Mittwoch,Erwin,Paula,Ben & Nele,, +2025-11-28,Donnerstag,Ben & Nele,Laura,Erwin,, +2025-11-29,Freitag,Paula,Erwin,Laura,Laura,Ben & Nele Erwin 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]