fairness ueber alle Dienste und Debugausgabe

This commit is contained in:
Jan Hoheisel 2025-12-24 21:56:50 +01:00
parent b885388122
commit 95b21aa150

View File

@ -106,8 +106,12 @@ class Elterndienstplaner:
return 0
def berechne_faire_zielverteilung_global(self) -> DefaultDict[str, DefaultDict[Dienst, float]]:
"""Berechnet die faire Zielanzahl von Diensten pro Eltern-Dienst-Kombination
basierend auf tatsächlich geleisteten historischen Diensten und deren fairer Umverteilung"""
"""Berechnet die faire Zielanzahl von Diensten für den Planungszeitraum
basierend auf globaler Fairness (Historie + aktueller Monat).
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: DefaultDict[str, DefaultDict[Dienst, float]] = \
defaultdict(lambda: defaultdict(float))
@ -186,19 +190,10 @@ class Elterndienstplaner:
faire_zuteilung = anteil * dienst.personen_anzahl
ziel_dienste[eltern][dienst] += faire_zuteilung
# Debug-Ausgabe für diesen Dienst
total_historisch = sum(ziel_dienste[e][dienst] for e in self.eltern) - benoetigte_dienste_monat
print(f" {dienst.kuerzel}: Historisch faire Summe={total_historisch:.1f}, "
f"Aktuell benötigt={benoetigte_dienste_monat}")
# Debug-Output: Detaillierte Zielverteilung
print("\n Berechnete Zielverteilung (basierend auf tatsächlichen historischen Diensten):")
for eltern in sorted(self.eltern):
for dienst in self.dienste:
if ziel_dienste[eltern][dienst] > 0.1: # Nur relevante Werte
ist = self.vorherige_dienste[eltern][dienst]
ziel = ziel_dienste[eltern][dienst]
print(f" {eltern} {dienst.kuerzel}: IST={ist}, FAIRE_ZIEL={ziel:.2f}, DIFF={ziel-ist:.2f}")
# 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:
ziel_dienste[eltern][dienst] -= self.vorherige_dienste[eltern][dienst]
return ziel_dienste
@ -245,10 +240,6 @@ class Elterndienstplaner:
faire_zuteilung = anteil * benoetigte_dienste_monat
ziel_dienste_lokal[eltern][dienst] = faire_zuteilung
if faire_zuteilung > 0.1: # Debug nur für relevante Werte
print(f" {eltern}: Faktor={monatlicher_dienstfaktor:.1f} "
f"-> {faire_zuteilung:.2f} Dienste")
return ziel_dienste_lokal
def _erstelle_entscheidungsvariablen(self) -> Dict[Tuple[str, date, Dienst], pulp.LpVariable]:
@ -363,20 +354,28 @@ class Elterndienstplaner:
self,
prob: pulp.LpProblem,
x: Dict[Tuple[str, date, Dienst], pulp.LpVariable],
ziel_dienste_global: DefaultDict[str, DefaultDict[Dienst, float]],
ziel_dienste_lokal: DefaultDict[str, DefaultDict[Dienst, float]]
) -> Tuple[Dict, Dict]:
"""F1 & F2: Erstellt Fairness-Variablen und fügt Fairness-Constraints hinzu (global und lokal)"""
ziel_dienste: DefaultDict[str, DefaultDict[Dienst, float]],
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_lokal = {} # F2
fairness_abweichung_global = {} # F1
fairness_abweichung = {}
for eltern in self.eltern:
for dienst in self.dienste:
fairness_abweichung_lokal[eltern, dienst] = pulp.LpVariable(
f"fair_lokal_{eltern.replace(' ', '_')}_{dienst.kuerzel}", lowBound=0)
fairness_abweichung_global[eltern, dienst] = pulp.LpVariable(
f"fair_global_{eltern.replace(' ', '_')}_{dienst.kuerzel}", lowBound=0)
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:
@ -388,57 +387,104 @@ class Elterndienstplaner:
if (eltern, tag, dienst) in x
)
# F2: Lokale Fairness - nur aktueller Monat
ziel_lokal = ziel_dienste_lokal[eltern][dienst]
if ziel_lokal > 0:
prob += (tatsaechliche_dienste_monat - ziel_lokal <=
fairness_abweichung_lokal[eltern, dienst])
prob += (ziel_lokal - tatsaechliche_dienste_monat <=
fairness_abweichung_lokal[eltern, dienst])
# Ziel für diese Fairness-Variante
ziel = ziel_dienste[eltern][dienst]
prob += (tatsaechliche_dienste_monat - ziel <=
fairness_abweichung[eltern, dienst])
prob += (ziel - tatsaechliche_dienste_monat <=
fairness_abweichung[eltern, dienst])
# F1: Globale Fairness - basierend auf berechneter Zielverteilung
ziel_global = ziel_dienste_global[eltern][dienst]
vorherige_dienste = self.vorherige_dienste[eltern][dienst]
return fairness_abweichung
if ziel_global > 0:
# Tatsächliche Dienste global (Vergangenheit + geplant)
total_dienste_inkl_vergangenheit = tatsaechliche_dienste_monat + vorherige_dienste
def _add_constraint_gesamtfairness(
self,
prob: pulp.LpProblem,
x: Dict[Tuple[str, date, Dienst], pulp.LpVariable],
ziel_dienste: DefaultDict[str, DefaultDict[Dienst, float]],
constraint_prefix: str
) -> Dict:
"""F3: Dienstübergreifende Fairness - verhindert Häufung bei einzelnen Eltern
prob += (total_dienste_inkl_vergangenheit - ziel_global <=
fairness_abweichung_global[eltern, dienst])
prob += (ziel_global - total_dienste_inkl_vergangenheit <=
fairness_abweichung_global[eltern, dienst])
Berechnet die Abweichung der Gesamtdienstanzahl (über alle Diensttypen)
vom fairen Gesamtziel. Dies verhindert, dass einzelne Eltern über alle
Diensttypen hinweg überproportional viele Dienste bekommen.
return fairness_abweichung_lokal, fairness_abweichung_global
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')
def _berechne_fairness_gewichte(self) -> Tuple[int, int]:
Returns:
Dictionary mit Gesamt-Fairness-Abweichungsvariablen
"""
fairness_abweichung_gesamt = {}
for eltern in self.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.tage
for dienst in self.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)
# 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 _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
if 9 <= aktueller_monat <= 12: # Sep-Dez: Jahresanfang
gewicht_f1 = 100 # Global wichtiger
gewicht_f2 = 50 # Lokal weniger wichtig
gewicht_f3_global = 30 # Gesamtfairness global
gewicht_f3_lokal = 15 # Gesamtfairness lokal
elif 1 <= aktueller_monat <= 3: # Jan-Mar: Jahresmitte
gewicht_f1 = 75
gewicht_f2 = 75
gewicht_f3_global = 25
gewicht_f3_lokal = 25
else: # Apr-Jul: Jahresende
gewicht_f1 = 50 # Global weniger wichtig
gewicht_f2 = 100 # Lokal wichtiger
gewicht_f3_global = 15
gewicht_f3_lokal = 30
return gewicht_f1, gewicht_f2
return gewicht_f1, gewicht_f2, gewicht_f3_global, gewicht_f3_lokal
def _erstelle_zielfunktion(
self,
prob: pulp.LpProblem,
x: Dict[Tuple[str, date, Dienst], pulp.LpVariable],
fairness_abweichung_lokal: Dict,
fairness_abweichung_global: 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_f1, gewicht_f2 = self._berechne_fairness_gewichte()
gewicht_global = 10
gewicht_lokal = 50
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.eltern:
@ -446,6 +492,10 @@ class Elterndienstplaner:
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.präferenzen.items():
if (eltern, tag, dienst) in x and präf == 1: # bevorzugt
@ -463,10 +513,20 @@ class Elterndienstplaner:
# 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}")
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, Dict[Tuple[str, date, Dienst], pulp.LpVariable]]:
"""Erstellt das PuLP Optimierungsmodell"""
def erstelle_optimierungsmodell(self) -> Tuple[
pulp.LpProblem,
Dict[Tuple[str, date, Dienst], pulp.LpVariable],
DefaultDict[str, DefaultDict[Dienst, float]],
DefaultDict[str, DefaultDict[Dienst, float]]
]:
"""Erstellt das PuLP Optimierungsmodell
Returns:
Tuple mit (prob, x, ziel_dienste_lokal, ziel_dienste_global)
"""
print("Erstelle Optimierungsmodell...")
# Debugging: Verfügbarkeit prüfen
@ -492,15 +552,32 @@ class Elterndienstplaner:
ziel_dienste_global = self.berechne_faire_zielverteilung_global()
ziel_dienste_lokal = self.berechne_faire_zielverteilung_lokal()
fairness_abweichung_lokal, fairness_abweichung_global = self._add_fairness_constraints(
prob, x, ziel_dienste_global, ziel_dienste_lokal
# 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)
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
return prob, x, ziel_dienste_lokal, ziel_dienste_global
def löse_optimierung(self, prob: pulp.LpProblem,
x: Dict[Tuple[str, date, Dienst], pulp.LpVariable]) -> Optional[Dict[date, Dict[Dienst, List[str]]]]:
@ -572,6 +649,87 @@ class Elterndienstplaner:
faktor_summe = sum(self.dienstfaktoren[eltern][tag] for tag in self.tage)
print(f" {eltern:15} {faktor_summe:.1f}")
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 Monat)
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:
@ -596,13 +754,17 @@ def main() -> None:
planer.lade_vorherige_ausgaben_csv(vorherige_datei)
# Optimierung
prob, x = planer.erstelle_optimierungsmodell()
prob, x, ziel_lokal, ziel_global = 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)
# Visualisierung der Verteilungen
planer.visualisiere_verteilungen(lösung, ziel_lokal, ziel_global)
print("\n✓ Planung erfolgreich abgeschlossen!")
else:
print("\n✗ Fehler: Keine gültige Lösung gefunden!")