redundante daten der dienstfaktoren entfernt

This commit is contained in:
Jan Hoheisel 2025-12-25 19:45:36 +01:00
parent 367e3cd316
commit 909e8ff9a0
2 changed files with 42 additions and 64 deletions

View File

@ -5,7 +5,7 @@ Trennt CSV-Parsing und -Schreiben von der Business-Logik
"""
import csv
from datetime import datetime, date
from datetime import datetime, date, timedelta
from typing import Dict, List, Tuple, DefaultDict
from collections import defaultdict
@ -97,24 +97,20 @@ class EingabeParser:
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
]:
def parse_eltern_csv(datei: str) -> Dict[str, DefaultDict[date, float]]:
"""
Lädt die eltern.csv mit Dienstfaktoren
Args:
datei: Pfad zur eltern.csv
tage: Liste der Planungstage
Returns:
Tuple mit (dienstfaktoren, alle_zeitraeume)
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 = {}
alle_zeitraeume = {}
with open(datei, 'r', encoding='utf-8') as f:
reader = csv.reader(f)
@ -126,9 +122,8 @@ class EingabeParser:
eltern_name = row[0].strip()
# Initialisiere Datenstrukturen
dienstfaktoren[eltern_name] = {}
alle_zeitraeume[eltern_name] = []
# 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):
@ -138,26 +133,18 @@ class EingabeParser:
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
# 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
# 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
return dienstfaktoren
@staticmethod
def parse_vorherige_ausgaben_csv(

View File

@ -46,7 +46,7 @@ class Elterndienstplaner:
]
# Datenstrukturen
self.tage: List[date] = []
self.planungszeitraum: List[date] = []
self.eltern: List[str] = []
self.benoetigte_dienste: Dict[date, List[Dienst]] = {}
self.verfügbarkeit: Dict[Tuple[str, date], bool] = {}
@ -56,7 +56,6 @@ class Elterndienstplaner:
# Wenn es eltern nicht gibt -> keyerror
# Wenn es tag nicht gibt -> default 0.0
self.dienstfaktoren: Dict[str, DefaultDict[date, float]] = {}
self.alle_zeitraeume: Dict[str, List[Tuple[date, date, float]]] = {}
self.historische_dienste: List[Tuple[date, str, Dienst]] = []
def get_dienst(self, kuerzel: str) -> Optional[Dienst]:
@ -80,13 +79,12 @@ class Elterndienstplaner:
def lade_eingabe_csv(self, datei: str) -> None:
"""Lädt die eingabe.csv mit Terminen und Präferenzen"""
self.eltern, self.tage, self.benoetigte_dienste, self.verfügbarkeit, self.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, self.alle_zeitraeume = \
EingabeParser.parse_eltern_csv(datei, self.tage)
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"""
@ -95,13 +93,9 @@ class Elterndienstplaner:
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.alle_zeitraeume:
return 0
for beginn, ende, faktor in self.alle_zeitraeume[eltern]:
if beginn <= datum <= ende:
return faktor
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]]:
"""Berechnet die faire Zielanzahl von Diensten für den Planungszeitraum
@ -137,35 +131,32 @@ class Elterndienstplaner:
dienste_pro_tag[datum].append(eltern)
# Für jeden historischen Tag faire Umverteilung berechnen
for hist_datum, geleistete_eltern in dienste_pro_tag.items():
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
dienstfaktoren_tag = {}
gesamt_dienstfaktor_tag = 0
for eltern in self.eltern:
faktor = self.berechne_dienstfaktor_an_datum(eltern, hist_datum)
dienstfaktoren_tag[eltern] = faktor
gesamt_dienstfaktor_tag += faktor
gesamt_dienstfaktor_tag += self.dienstfaktoren[eltern][tag]
# Faire Umverteilung der an diesem Tag geleisteten Dienste
if gesamt_dienstfaktor_tag > 0:
for eltern in self.eltern:
if dienstfaktoren_tag[eltern] > 0:
anteil = dienstfaktoren_tag[eltern] / gesamt_dienstfaktor_tag
if self.dienstfaktoren[eltern][tag] > 0:
anteil = self.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" {hist_datum}: {eltern} Faktor={dienstfaktoren_tag[eltern]} "
print(f" {tag}: {eltern} Faktor={self.dienstfaktoren[eltern][tag]} "
f"-> {faire_zuteilung:.2f} von {anzahl_dienste} Diensten")
# 2. AKTUELLER MONAT: Faire Verteilung der benötigten Dienste (tageweise wie bei historischen Diensten)
benoetigte_dienste_monat = 0
# Für jeden Tag im aktuellen Monat faire Umverteilung berechnen
for tag in self.tage:
for tag in self.planungszeitraum:
# Prüfe ob an diesem Tag der Dienst benötigt wird
if dienst not in self.benoetigte_dienste.get(tag, []):
continue
@ -173,18 +164,18 @@ class Elterndienstplaner:
benoetigte_dienste_monat += dienst.personen_anzahl
# Dienstfaktoren aller Eltern für diesen Tag berechnen
dienstfaktoren_tag = {}
dienstfaktoren = {}
gesamt_dienstfaktor_tag = 0
for eltern in self.eltern:
faktor = self.dienstfaktoren[eltern][tag]
dienstfaktoren_tag[eltern] = faktor
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:
anteil = dienstfaktoren_tag[eltern] / gesamt_dienstfaktor_tag
anteil = dienstfaktoren[eltern] / gesamt_dienstfaktor_tag
faire_zuteilung = anteil * dienst.personen_anzahl
ziel_dienste[eltern][dienst] += faire_zuteilung
@ -211,7 +202,7 @@ class Elterndienstplaner:
# Gesamtdienstfaktor für aktuellen Monat berechnen
gesamt_dienstfaktor_monat = sum(
sum(self.dienstfaktoren[e][tag] for tag in self.tage)
sum(self.dienstfaktoren[e][tag] for tag in self.planungszeitraum)
for e in self.eltern
)
@ -223,7 +214,7 @@ class Elterndienstplaner:
for dienst in self.dienste:
# Anzahl benötigter Dienste im aktuellen Monat
benoetigte_dienste_monat = sum(
1 for tag in self.tage
1 for tag in self.planungszeitraum
if dienst in self.benoetigte_dienste.get(tag, [])
)
# Multipliziere mit Anzahl benötigter Personen pro Dienst
@ -235,7 +226,7 @@ class Elterndienstplaner:
for eltern in self.eltern:
# Dienstfaktor für diesen Elternteil im aktuellen Monat
monatlicher_dienstfaktor = sum(
self.dienstfaktoren[eltern][tag] for tag in self.tage
self.dienstfaktoren[eltern][tag] for tag in self.planungszeitraum
)
if monatlicher_dienstfaktor > 0:
@ -249,7 +240,7 @@ class Elterndienstplaner:
"""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.tage:
for tag in self.planungszeitraum:
for dienst in self.dienste:
if dienst in self.benoetigte_dienste.get(tag, []):
x[eltern, tag, dienst] = pulp.LpVariable(
@ -264,13 +255,13 @@ class Elterndienstplaner:
x: Dict[Tuple[str, date, Dienst], pulp.LpVariable]
) -> None:
"""C1: Je Eltern und Dienst nur einmal die Woche (Woche = Montag bis Sonntag)"""
erster_tag = self.tage[0]
erster_tag = self.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.tage[-1]
letzter_tag = self.planungszeitraum[-1]
while woche_start <= letzter_tag:
woche_ende = woche_start + timedelta(days=6) # Sonntag
@ -289,7 +280,7 @@ class Elterndienstplaner:
historische_dienste_in_woche += 1
# Sammle Variablen für Planungszeitraum in dieser Woche
for tag in self.tage:
for tag in self.planungszeitraum:
if woche_start <= tag <= woche_ende:
if (eltern, tag, dienst) in x:
woche_vars.append(x[eltern, tag, dienst])
@ -309,7 +300,7 @@ class Elterndienstplaner:
) -> None:
"""C2: Je Eltern nur einen Dienst am Tag"""
for eltern in self.eltern:
for tag in self.tage:
for tag in self.planungszeitraum:
tag_vars = []
for dienst in self.dienste:
if (eltern, tag, dienst) in x:
@ -325,7 +316,7 @@ class Elterndienstplaner:
) -> None:
"""C3: Dienste nur verfügbaren Eltern zuteilen"""
for eltern in self.eltern:
for tag in self.tage:
for tag in self.planungszeitraum:
if not self.verfügbarkeit.get((eltern, tag), True):
for dienst in self.dienste:
if (eltern, tag, dienst) in x:
@ -338,7 +329,7 @@ class Elterndienstplaner:
x: Dict[Tuple[str, date, Dienst], pulp.LpVariable]
) -> None:
"""C4: Alle benötigten Dienste müssen zugeteilt werden"""
for tag in self.tage:
for tag in self.planungszeitraum:
for dienst in self.benoetigte_dienste.get(tag, []):
dienst_vars = []
for eltern in self.eltern:
@ -386,7 +377,7 @@ class Elterndienstplaner:
# Tatsächliche Dienste im aktuellen Monat
tatsaechliche_dienste_monat = pulp.lpSum(
x[eltern, tag, dienst]
for tag in self.tage
for tag in self.planungszeitraum
if (eltern, tag, dienst) in x
)
@ -431,7 +422,7 @@ class Elterndienstplaner:
# Tatsächliche Gesamtdienste für diesen Elternteil
tatsaechliche_dienste_gesamt = pulp.lpSum(
x[eltern, tag, dienst]
for tag in self.tage
for tag in self.planungszeitraum
for dienst in self.dienste
if (eltern, tag, dienst) in x
)
@ -449,7 +440,7 @@ class Elterndienstplaner:
def _berechne_fairness_gewichte(self) -> Tuple[int, int, int, int]:
"""Berechnet Gewichtung basierend auf Jahreszeit (Sep-Jul Schuljahr)"""
aktueller_monat = self.tage[0].month if self.tage else 1
aktueller_monat = self.planungszeitraum[0].month if self.planungszeitraum else 1
if 9 <= aktueller_monat <= 12: # Sep-Dez: Jahresanfang
gewicht_f1 = 100 # Global wichtiger
@ -534,7 +525,7 @@ class Elterndienstplaner:
# Debugging: Verfügbarkeit prüfen
print("\nDebug: Verfügbarkeit analysieren...")
for tag in self.tage[:5]: # Erste 5 Tage
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, [])
print(f" {tag}: Benötigt {len(benötigte)} Dienste {benötigte}, verfügbar: {verfügbare}")
@ -623,7 +614,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"""
AusgabeWriter.schreibe_ausgabe_csv(datei, lösung, self.tage, self.dienste)
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"""
@ -649,7 +640,7 @@ class Elterndienstplaner:
# Dienstfaktor-Analyse
print(f"\nDienstfaktoren im Planungszeitraum:")
for eltern in sorted(self.eltern):
faktor_summe = sum(self.dienstfaktoren[eltern][tag] for tag in self.tage)
faktor_summe = sum(self.dienstfaktoren[eltern][tag] for tag in self.planungszeitraum)
print(f" {eltern:15} {faktor_summe:.1f}")
def visualisiere_praeferenz_verletzungen(