diff --git a/STRUKTUR.md b/STRUKTUR.md new file mode 100644 index 0000000..ad7db20 --- /dev/null +++ b/STRUKTUR.md @@ -0,0 +1,98 @@ +# 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 + - Statistiken + +**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 +└── CSV Schreiben +``` + +### Nachher (Modular) +``` +csv_io.py (220 Zeilen) +├── EingabeParser +└── AusgabeWriter + +elterndienstplaner.py (500 Zeilen) +├── Dienst Klasse +├── Fairness-Berechnung +├── Optimierung +└── Statistiken +``` + +## Nächste Schritte (Optional) + +### Phase 2: Constraint-Funktionen auslagern +```python +def _add_constraint_ein_dienst_pro_woche(prob, x, ...): + """C1: Je Eltern und Dienst nur einmal die Woche""" + +def _add_constraint_ein_dienst_pro_tag(prob, x, ...): + """C2: Je Eltern nur einen Dienst am Tag""" +``` + +### 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/csv_io.py b/csv_io.py new file mode 100644 index 0000000..f02e35f --- /dev/null +++ b/csv_io.py @@ -0,0 +1,304 @@ +#!/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 +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 + dienst_objekte = [] + for kuerzel in dienste_str: + dienst = dienste_lookup(kuerzel) + if dienst: + dienst_objekte.append(dienst) + benoetigte_dienste[datum_obj] = dienst_objekte + + # 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, tage: List[date]) -> Tuple[ + Dict[str, Dict[date, float]], # dienstfaktoren + Dict[str, List[Tuple[date, date, float]]] # alle_zeitraeume + ]: + """ + Lädt die eltern.csv mit Dienstfaktoren + + Args: + datei: Pfad zur eltern.csv + tage: Liste der Planungstage + + Returns: + Tuple mit (dienstfaktoren, alle_zeitraeume) + """ + print(f"Lade Elterndaten aus {datei}...") + + dienstfaktoren = {} + alle_zeitraeume = {} + + 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 Datenstrukturen + dienstfaktoren[eltern_name] = {} + alle_zeitraeume[eltern_name] = [] + + # 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 + + # Zeitraum speichern + alle_zeitraeume[eltern_name].append((beginn, ende, faktor)) + + # Faktor für Tage im aktuellen Planungsmonat setzen + for tag in tage: + if beginn <= tag <= ende: + dienstfaktoren[eltern_name][tag] = faktor + + except (ValueError, IndexError): + continue + + # Tage ohne expliziten Faktor auf 0 setzen + for tag in tage: + if tag not in dienstfaktoren[eltern_name]: + dienstfaktoren[eltern_name][tag] = 0 + + print(f"Dienstfaktoren geladen für {len(dienstfaktoren)} Eltern") + print(f"Zeiträume gespeichert für globale Fairness-Berechnung") + + return dienstfaktoren, alle_zeitraeume + + @staticmethod + def parse_vorherige_ausgaben_csv( + datei: str, + eltern: List[str], + dienste: List, # List[Dienst] + ) -> Tuple[ + DefaultDict[str, DefaultDict], # vorherige_dienste + 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: + Tuple mit (vorherige_dienste, historische_dienste) + """ + print(f"Lade vorherige Ausgaben aus {datei}...") + + vorherige_dienste = defaultdict(lambda: defaultdict(int)) + 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 zählen UND 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 Leerzeichen getrennt) + eltern_liste = row[spalte_idx].strip().split() + for eltern_name in eltern_liste: + if eltern_name in eltern: + # Summierung für Kompatibilität + vorherige_dienste[eltern_name][dienst] += 1 + # 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"Vorherige Dienste geladen: {dict(vorherige_dienste)}") + print(f"Historische Dienste mit Datum: {len(historische_dienste)} Einträge") + + return vorherige_dienste, 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 = ' '.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/elterndienstplaner.py b/elterndienstplaner.py index d72d489..3e0ffda 100755 --- a/elterndienstplaner.py +++ b/elterndienstplaner.py @@ -7,12 +7,12 @@ Datum: Dezember 2025 """ import sys -import csv import pulp -from datetime import datetime, timedelta, date +from datetime import timedelta, date from collections import defaultdict from typing import Dict, List, Tuple, DefaultDict, Optional -import calendar + +from csv_io import EingabeParser, AusgabeWriter class Dienst: @@ -78,169 +78,18 @@ class Elterndienstplaner: def lade_eingabe_csv(self, datei: str) -> None: """Lädt die eingabe.csv mit Terminen und Präferenzen""" - print(f"Lade Eingabedaten aus {datei}...") - - with open(datei, 'r', encoding='utf-8') as f: - reader = csv.reader(f) - header = next(reader) - - # Eltern aus Header extrahieren (ab Spalte 3) - self.eltern = [name.strip() for name in header[3:] if name.strip()] - print(f"Gefundene Eltern: {self.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() - self.tage.append(datum_obj) - except ValueError: - print(f"Warnung: Ungültiges Datum {datum}") - continue - - # Benötigte Dienste - dienst_objekte = [] - for kuerzel in dienste_str: - dienst = self.get_dienst(kuerzel) - if dienst: - dienst_objekte.append(dienst) - self.benoetigte_dienste[datum_obj] = dienst_objekte - - # Verfügbarkeit und Präferenzen der Eltern - for i, eltern_name in enumerate(self.eltern): - if i + 3 < len(row): - präf_str = row[i + 3].strip() - - # Verfügbarkeit prüfen - if präf_str == 'x': - self.verfügbarkeit[(eltern_name, datum_obj)] = False - else: - self.verfügbarkeit[(eltern_name, datum_obj)] = True - - # Präferenzen parsen - self._parse_präferenzen(eltern_name, datum_obj, präf_str) - else: - # Standard: verfügbar, keine Präferenzen - self.verfügbarkeit[(eltern_name, datum_obj)] = True - - self.tage.sort() - print(f"Zeitraum: {self.tage[0]} bis {self.tage[-1]} ({len(self.tage)} Tage)") - - def _parse_präferenzen(self, eltern: str, datum: date, präf_str: str) -> None: - """Parst Präferenzstring wie 'F+P-E+' """ - i = 0 - while i < len(präf_str): - if i + 1 < len(präf_str): - dienst = self.get_dienst(präf_str[i]) - if dienst and i + 1 < len(präf_str): - if präf_str[i + 1] == '+': - self.präferenzen[(eltern, datum, dienst)] = 1 - i += 2 - elif präf_str[i + 1] == '-': - self.präferenzen[(eltern, datum, dienst)] = -1 - i += 2 - else: - i += 1 - else: - i += 1 - else: - i += 1 + self.eltern, self.tage, 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""" - print(f"Lade Elterndaten aus {datei}...") - - 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 Datenstrukturen - self.dienstfaktoren[eltern_name] = {} - self.alle_zeitraeume[eltern_name] = [] - - # 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 - - # Zeitraum speichern - self.alle_zeitraeume[eltern_name].append((beginn, ende, faktor)) - - # Faktor für Tage im aktuellen Planungsmonat setzen - for tag in self.tage: - if beginn <= tag <= ende: - self.dienstfaktoren[eltern_name][tag] = faktor - - except (ValueError, IndexError): - continue - - # Tage ohne expliziten Faktor auf 0 setzen - for tag in self.tage: - if tag not in self.dienstfaktoren[eltern_name]: - self.dienstfaktoren[eltern_name][tag] = 0 - - print(f"Dienstfaktoren geladen für {len(self.dienstfaktoren)} Eltern") - print(f"Zeiträume gespeichert für globale Fairness-Berechnung") + self.dienstfaktoren, self.alle_zeitraeume = \ + EingabeParser.parse_eltern_csv(datei, self.tage) def lade_vorherige_ausgaben_csv(self, datei: str) -> None: """Lädt vorherige-ausgaben.csv für Fairness-Constraints""" - print(f"Lade vorherige Ausgaben aus {datei}...") - - 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 self.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 zählen UND 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 Leerzeichen getrennt) - eltern_liste = row[spalte_idx].strip().split() - for eltern_name in eltern_liste: - if eltern_name in self.eltern: - # Summierung für Kompatibilität - self.vorherige_dienste[eltern_name][dienst] += 1 - # Historische Dienste mit Datum speichern - self.historische_dienste.append((datum, eltern_name, dienst)) - - except FileNotFoundError: - print("Keine vorherigen Ausgaben gefunden - starte ohne historische Daten") - - print(f"Vorherige Dienste geladen: {dict(self.vorherige_dienste)}") - print(f"Historische Dienste mit Datum: {len(self.historische_dienste)} Einträge") + self.vorherige_dienste, 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""" @@ -620,31 +469,7 @@ class Elterndienstplaner: def schreibe_ausgabe_csv(self, datei: str, lösung: Dict[date, Dict[Dienst, List[str]]]) -> None: """Schreibt die Lösung in die ausgabe.csv""" - 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 self.dienste] - writer.writerow(header) - - # Daten schreiben - for tag in sorted(self.tage): - wochentag = ['Montag', 'Dienstag', 'Mittwoch', 'Donnerstag', 'Freitag', 'Samstag', 'Sonntag'][tag.weekday()] - - row = [tag.strftime('%Y-%m-%d'), wochentag] - - for dienst in self.dienste: - if tag in lösung and dienst in lösung[tag]: - eltern_str = ' '.join(lösung[tag][dienst]) - else: - eltern_str = '' - row.append(eltern_str) - - writer.writerow(row) - - print("Ausgabe erfolgreich geschrieben!") + AusgabeWriter.schreibe_ausgabe_csv(datei, lösung, self.tage, self.dienste) def drucke_statistiken(self, lösung: Dict[date, Dict[Dienst, List[str]]]) -> None: """Druckt Statistiken zur Lösung"""