#!/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, Tuple 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 # Historische Dienste (kann über Observer gesetzt werden) self.historische_dienste: List[Tuple[date, Eltern, Dienst]] = 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 setze_historische_dienste(self, historische_dienste: List[Tuple[date, Eltern, Dienst]]) -> None: """Observer-Callback: Setzt historische Dienste für Ausgabe/Export""" self.historische_dienste = historische_dienste 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, False) # Schreibe ergänzende Datei mit historischen Diensten (falls vorhanden). hist_datei = datei.replace('.csv', '-gesamt.csv') if datei.endswith('.csv') else datei + '-gesamt.csv' historische = self.historische_dienste if self.historische_dienste is not None else self.daten.historische_dienste AusgabeWriter.schreibe_ausgabe_csv(hist_datei, lösung, self.daten.planungszeitraum, self.daten.dienste, True, historische) 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.praeferenzen.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 for tag in zugeteilte_tage: if tag not in positive_praef_tage: verletzungen[eltern][dienst]['positiv_nicht_erfuellt'] += 1 # Tabelle ausgeben (verbesserte Spaltenformatierung) col_width = 14 # Breite pro Dienst-Spalte (sichtbar) name_col = 20 # Header print(f"\n{'Eltern':<{name_col}}", end='') for dienst in self.daten.dienste: print(f"{dienst.kuerzel:^{col_width}}", end='') print() print(f"{'':{name_col}}", end='') for _ in self.daten.dienste: print(f"{'neg, pos':^{col_width}}", end='') print() print("-" * (name_col + col_width * len(self.daten.dienste))) gesamt_negativ = defaultdict(int) gesamt_positiv = defaultdict(int) for eltern in sorted(self.daten.eltern): print(f"{eltern:<{name_col}}", 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 # Inhalt vor Padding erstellen cell = f"{neg:>3}, {pos:>3}" cell_padded = cell.center(col_width) # Farbcodierung (erst nach Padding anwenden) 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}{cell_padded}{reset}", end='') print() # Summenzeile print("-" * (name_col + col_width * len(self.daten.dienste))) print(f"{'SUMME':<{name_col}}", end='') for dienst in self.daten.dienste: neg = gesamt_negativ[dienst] pos = gesamt_positiv[dienst] cell = f"{neg:>3}, {pos:>3}" cell_padded = cell.center(col_width) farbe = "" reset = "" if neg > 0 or pos > 0: farbe = "\033[91m" if neg > 0 else "\033[93m" reset = "\033[0m" print(f"{farbe}{cell_padded}{reset}", 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)") def visualisiere_dienste_uebersicht( self, lösung: Dict[date, Dict[Dienst, List[Eltern]]] ) -> None: """Visualisiert die Übersicht der zugeteilten Dienste nach Optimierung Zeigt für jede Familie und jeden Diensttyp: - Anzahl Dienste nach Optimierung (Historie + Planungszeitraum) - Differenz zum globalen Ziel (Historie + Planungszeitraum) Args: lösung: Die Lösung der Optimierung """ if self.ziel_global is None: print("FEHLER: Globale Zielverteilung wurde nicht gesetzt!") return # Berechne historische Dienste pro Eltern und Dienst historisch = defaultdict(lambda: defaultdict(int)) for datum, eltern, dienst in self.daten.historische_dienste: historisch[eltern][dienst] += 1 # Berechne geplante Dienste (aus Lösung) geplant = defaultdict(lambda: defaultdict(int)) for tag_dienste in lösung.values(): for dienst, eltern_liste in tag_dienste.items(): for eltern in eltern_liste: geplant[eltern][dienst] += 1 # Berechne globale Ziele für jeden Elternteil und Dienst # Das globale Ziel ist: faire Verteilung über (Historie + Planungszeitraum) MINUS bereits geleistete Historie # Also: ziel_global[eltern][dienst] ist die SOLL-Änderung im Planungszeitraum # Tatsächliches Gesamt-Ziel = historisch[eltern][dienst] + ziel_global[eltern][dienst] print("\n" + "="*120) print("ÜBERSICHT: Dienst nach der Optimierung") print("="*120) # Tabelle: NACH der Optimierung (historisch + geplant) print("\n>>> NACH OPTIMIERUNG (historische Dienste + Planungszeitraum) <<<\n") # Header print(f"{'Eltern':<20} ", end='') for dienst in self.daten.dienste: print(f"{dienst.kuerzel:>14} ", end='') print(f"{'GESAMT':>14}") print(f"{'':20} ", end='') for dienst in self.daten.dienste: print(f"{'Ist / Δ Ziel':>14} ", end='') print(f"{'Ist / Δ Ziel':>14}") print("-" * 120) # Datenzeilen for eltern in sorted(self.daten.eltern): print(f"{eltern:<20} ", end='') gesamt_ist = 0 gesamt_ziel = 0 for dienst in self.daten.dienste: ist_dienste = historisch[eltern][dienst] + geplant[eltern][dienst] gesamt_ist += ist_dienste # Globales Ziel = historisch + ziel_global (das ist das faire Gesamt-Ziel) ziel_gesamt = historisch[eltern][dienst] + self.ziel_global[eltern][dienst] gesamt_ziel += ziel_gesamt delta = ist_dienste - ziel_gesamt # Farbcodierung farbe = "" reset = "" if abs(delta) > 0.5: farbe = "\033[93m" if abs(delta) <= 1.5 else "\033[91m" reset = "\033[0m" print(f"{farbe}{ist_dienste:>6} / {delta:>+5.1f}{reset} ", end='') # Gesamt-Spalte delta_gesamt = gesamt_ist - gesamt_ziel farbe = "" reset = "" if abs(delta_gesamt) > 0.5: farbe = "\033[93m" if abs(delta_gesamt) <= 1.5 else "\033[91m" reset = "\033[0m" print(f"{farbe}{gesamt_ist:>6} / {delta_gesamt:>+5.1f}{reset}") print() print("Legende:") print(" Ist = Anzahl tatsächlich geleisteter Dienste") print(" Δ Ziel = Differenz zum globalen fairen Ziel (positiv = mehr als fair, negativ = weniger)") print(" \033[93mGelb\033[0m = Abweichung 0.5 - 1.5 Dienste") print(" \033[91mRot\033[0m = Abweichung > 1.5 Dienste")