#!/usr/bin/env python3

# Infeasibility_Diagnose.py
"""
Kapitel Praxisfallen: Aus INFEASIBLE einen Notfallplan machen.

Demonstriert die hierarchische Relaxation an einem Dienstplan, der in der
harten Fassung unloesbar ist, und zeigt, wie der Solver die Ursache benennt.
"""

from ortools.sat.python import cp_model

SLOTS = ["Mo frueh", "Mo spaet", "Di frueh", "Di spaet"]
FAECHER = ["Mathematik", "Physik", "Mathematik", "Chemie"]   # Chemie kann niemand!
PERSONAL = ["Alice", "Bob", "Carla"]
QUALIFIKATION = {
    "Alice": {"Mathematik", "Physik"},
    "Bob":   {"Mathematik"},
    "Carla": {"Physik", "Mathematik"},
}
MAX_PRO_PERSON = 2
STRAFE_UNBESETZT = 10_000          # sehr teuer, aber nicht unmoeglich
STRAFE_UEBERLAST = 500             # teuer, aber billiger als Ausfall


def plane(harte_fassung: bool):
    """harte_fassung=True: klassisch (kann INFEASIBLE werden).
       harte_fassung=False: mit Schlupfvariablen (immer loesbar)."""
    modell = cp_model.CpModel()
    x = {(p, s): modell.NewBoolVar(f"x_{p}_{s}")
         for p in PERSONAL for s in range(len(SLOTS))}

    strafen = []
    unbesetzt = {}
    ueberlast = {}

    for s in range(len(SLOTS)):
        if harte_fassung:
            modell.AddExactlyOne(x[p, s] for p in PERSONAL)
        else:
            # Schlupfvariable: der Slot DARF unbesetzt bleiben - gegen hohe Strafe
            unbesetzt[s] = modell.NewBoolVar(f"unbesetzt_{s}")
            modell.AddExactlyOne([x[p, s] for p in PERSONAL] + [unbesetzt[s]])
            strafen.append(unbesetzt[s] * STRAFE_UNBESETZT)

    # Qualifikation bleibt IMMER hart - fachfremder Unterricht ist keine Option
    for s, fach in enumerate(FAECHER):
        for p in PERSONAL:
            if fach not in QUALIFIKATION[p]:
                modell.Add(x[p, s] == 0)

    for p in PERSONAL:
        last = sum(x[p, s] for s in range(len(SLOTS)))
        if harte_fassung:
            modell.Add(last <= MAX_PRO_PERSON)
        else:
            # Ueberlast erlaubt - aber teuer
            ueberlast[p] = modell.NewIntVar(0, len(SLOTS), f"ueberlast_{p}")
            modell.Add(last <= MAX_PRO_PERSON + ueberlast[p])
            strafen.append(ueberlast[p] * STRAFE_UEBERLAST)

    if strafen:
        modell.Minimize(sum(strafen))

    loeser = cp_model.CpSolver()
    loeser.parameters.max_time_in_seconds = 5.0
    status = loeser.Solve(modell)
    return loeser, status, x, unbesetzt, ueberlast


if __name__ == "__main__":
    print("=" * 78)
    print("  VARIANTE A: KLASSISCH MIT LAUTER HARTEN BEDINGUNGEN")
    print("=" * 78)
    loeser, status, *_ = plane(harte_fassung=True)
    print(f"Solver-Status: {loeser.StatusName(status)}")
    if status == cp_model.INFEASIBLE:
        print("Das System kann keinen Plan liefern. Der Anwender erfaehrt NICHT,")
        print("welche Regel das Problem verursacht - nur, dass es nicht geht.\n")

    print("=" * 78)
    print("  VARIANTE B: MIT HIERARCHISCHER RELAXATION")
    print("=" * 78)
    loeser, status, x, unbesetzt, ueberlast = plane(harte_fassung=False)
    print(f"Solver-Status: {loeser.StatusName(status)} | "
          f"Strafkosten: {loeser.ObjectiveValue():.0f}\n")

    print(f"{'Slot':<12} {'Fach':<12} {'Zuweisung':<18} {'Bemerkung'}")
    print("-" * 78)
    for s, slot in enumerate(SLOTS):
        if loeser.Value(unbesetzt[s]):
            print(f"{slot:<12} {FAECHER[s]:<12} {'-- UNBESETZT --':<18} "
                  f"Kein qualifiziertes Personal verfuegbar")
        else:
            person = next(p for p in PERSONAL if loeser.Value(x[p, s]))
            print(f"{slot:<12} {FAECHER[s]:<12} {person:<18}")

    print("\n--- Diagnose ---")
    for s, slot in enumerate(SLOTS):
        if loeser.Value(unbesetzt[s]):
            fach = FAECHER[s]
            qualifiziert = [p for p in PERSONAL if fach in QUALIFIKATION[p]]
            print(f"  '{slot}' ({fach}): {len(qualifiziert)} qualifizierte Personen "
                  f"{qualifiziert if qualifiziert else '-> URSACHE: niemand kann dieses Fach'}")
    for p in PERSONAL:
        if loeser.Value(ueberlast[p]):
            print(f"  {p} arbeitet {loeser.Value(ueberlast[p])} Stunden ueber der Grenze.")

    print("\nDer Anwender bekommt jetzt einen Plan PLUS eine konkrete Ursache -")
    print("und kann handeln: Vertretung von aussen holen, Stunde verlegen,")
    print("oder die Klasse zusammenlegen.")
    print("=" * 78)

#!/usr/bin/env python3

# Erklaerbarkeit.py
"""
Kapitel Praxisfallen: Ein Constraint-Trace, der die drei Fragen der Faustregel fuer
Erklaerbarkeit (Abschnitt 'Die fuenf typischen Praxisfallen') fuer eine
einzelne Zuweisung beantwortet:
Warum diese Loesung? Warum nicht die Alternative? Was wuerde sie verbessern?
"""

from ortools.sat.python import cp_model

PERSONAL = ["Anna", "Ben", "Clara"]
BEREITS_STUNDEN = {"Anna": 1, "Ben": 2, "Clara": 0}   # diese Woche schon geleistet
MAX_STUNDEN = 2                                        # gesetzliche Obergrenze
STRAFE_VORBELASTUNG = 50                                # pro bereits geleisteter Stunde
STRAFKOSTEN_WUNSCH = {"Anna": 0, "Ben": 30, "Clara": 100}


def loese(max_stunden: int):
    modell = cp_model.CpModel()
    x = {p: modell.NewBoolVar(f"x_{p}") for p in PERSONAL}
    modell.AddExactlyOne(x.values())

    kosten = []
    for p in PERSONAL:
        if BEREITS_STUNDEN[p] + 1 > max_stunden:
            modell.Add(x[p] == 0)                       # hartes Ausschlusskriterium
        strafe = BEREITS_STUNDEN[p] * STRAFE_VORBELASTUNG + STRAFKOSTEN_WUNSCH[p]
        kosten.append(strafe * x[p])

    modell.Minimize(sum(kosten))
    loeser = cp_model.CpSolver()
    status = loeser.Solve(modell)
    if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):
        return None
    gewinner = next(p for p in PERSONAL if loeser.Value(x[p]))
    return gewinner, loeser.ObjectiveValue()


def gesamtkosten(p: str) -> int:
    return BEREITS_STUNDEN[p] * STRAFE_VORBELASTUNG + STRAFKOSTEN_WUNSCH[p]


if __name__ == "__main__":
    gewinner, kosten_opt = loese(MAX_STUNDEN)

    print("=" * 78)
    print("  CONSTRAINT-TRACE: Warum bekommt", gewinner, "die Vertretungsstunde?")
    print("=" * 78)

    print("\n1. WARUM DIESE LOESUNG?")
    v, w = BEREITS_STUNDEN[gewinner] * STRAFE_VORBELASTUNG, STRAFKOSTEN_WUNSCH[gewinner]
    print(f"   {gewinner}: Vorbelastung {v} Strafpunkte + Wunsch {w} Strafpunkte "
          f"= {v + w} Strafpunkte insgesamt (Minimum).")

    print("\n2. WARUM NICHT DIE NAHELIEGENDEN ALTERNATIVEN?")
    for p in PERSONAL:
        if p == gewinner:
            continue
        if BEREITS_STUNDEN[p] + 1 > MAX_STUNDEN:
            print(f"   {p}: AUSGESCHLOSSEN - hat bereits {BEREITS_STUNDEN[p]} Stunden, "
                  f"eine dritte würde die Obergrenze von {MAX_STUNDEN} Stunden verletzen "
                  f"(harte Bedingung, nicht verhandelbar).")
        else:
            differenz = gesamtkosten(p) - kosten_opt
            print(f"   {p}: zulässig, aber {differenz:.0f} Strafpunkte teurer als "
                  f"{gewinner} ({gesamtkosten(p):.0f} statt {kosten_opt:.0f}).")

    print("\n3. WAS WUERDE DIE LOESUNG VERBESSERN?")
    gewinner_gelockert, kosten_gelockert = loese(MAX_STUNDEN + 1)
    if kosten_gelockert < kosten_opt:
        print(f"   Obergrenze auf {MAX_STUNDEN + 1} Stunden gelockert: {gewinner_gelockert} "
              f"uebernimmt jetzt zu {kosten_gelockert:.0f} Strafpunkten "
              f"({kosten_opt - kosten_gelockert:.0f} weniger als heute).")
        print("   -> Diese Bedingung hat tatsaechlich einen Preis.")
    else:
        print(f"   Obergrenze auf {MAX_STUNDEN + 1} Stunden gelockert: Ergebnis bleibt bei "
              f"{gewinner} mit {kosten_gelockert:.0f} Strafpunkten (unveraendert).")
        blockiert = ", ".join(p for p in PERSONAL
                               if BEREITS_STUNDEN[p] + 1 > MAX_STUNDEN and p != gewinner)
        print(f"   -> Die Obergrenze blockiert zwar {blockiert}, kostet aber gerade NICHTS:")
        print("   Selbst ohne sie waere die Loesung dieselbe.")
        print("   Blockierend und kostenrelevant sind zwei verschiedene Dinge.")
    print("=" * 78)

#!/usr/bin/env python3

# or_kern.py
"""
Kapitel Praxisfallen: Der gemeinsame Unterbau fuer produktionsreife OR-Modelle.

Alle Beispiele dieses Buchs loesen dieselben vier Aufgaben immer wieder:
Daten einlesen und pruefen, ein Modell bauen, den Solverstatus auswerten, die
Loesung gegen die Wirklichkeit kontrollieren. Dieses Modul zieht diese vier
Aufgaben aus den Einzelprogrammen heraus.

    Rohdaten (Excel/CSV)
        -> Domaenenmodell (Pydantic, geprueft)
            -> Modellbauer (solverabhaengig)
                -> Loesung (DTO, solverunabhaengig)
                    -> Abnahmepruefung

Der Sinn dieser Trennung: Die Domaenenschicht weiss nichts von Solvern, die
Loesungsschicht weiss nichts vom Modellaufbau. Wer den Solver wechselt, tauscht
genau EINEN Baustein aus - der Rest bleibt.

WICHTIG ZUM IMPORT: Dieses Modul importiert von sich aus KEINE Solver-
bibliothek. Der Grund steht im Kapitel Oekosystem: ortools und highspy vertragen sich
nicht im selben Prozess. Die Statusuebersetzer laden ihre Bibliothek erst,
wenn sie aufgerufen werden - so bleibt or_kern.py mit jedem Solver benutzbar.

Benoetigt: pydantic (v2), numpy; optional pandas und openpyxl fuer Excel.
"""

from __future__ import annotations

from enum import Enum
from typing import Annotated, Any, Sequence

import numpy as np
from pydantic import BaseModel, Field, model_validator

# Wiederverwendbare Feldtypen. Sie tragen die Pruefung im Typ, nicht im Code -
# damit gilt sie ueberall, wo der Typ verwendet wird.
PositiveZahl = Annotated[float, Field(gt=0)]
NichtNegativ = Annotated[float, Field(ge=0)]


# --- 1. Solverstatus: eine Sprache fuer fuenf Bibliotheken -------------------

class SolverStatus(str, Enum):
    """Was ein Solverlauf ergeben hat - unabhaengig davon, wer gerechnet hat.

    Die fuenf im Buch verwendeten Bibliotheken benennen dasselbe Ergebnis
    unterschiedlich: 'Optimal', 'OPTIMAL', 2, 'optimal'. Wer darauf direkt
    prueft, bindet seinen Auswertungscode an eine Bibliothek.
    """
    OPTIMAL = "optimal"              # beweisbar bestmoeglich
    ZULAESSIG = "zulaessig"          # brauchbare Loesung, Beweis fehlt
    UNZULAESSIG = "unzulaessig"      # es gibt keine Loesung (Aussage ueber das Modell)
    UNBESCHRAENKT = "unbeschraenkt"  # Ziel waechst ins Unendliche
    ZEITLIMIT = "zeitlimit"          # abgebrochen, nichts Brauchbares gefunden
    FEHLERHAFT = "fehlerhaft"        # das Modell ist kein gueltiges Modell
    UNBEKANNT = "unbekannt"          # alles Uebrige

    @property
    def brauchbar(self) -> bool:
        """Habe ich etwas in der Hand, das ich ausfuehren kann?"""
        return self in (SolverStatus.OPTIMAL, SolverStatus.ZULAESSIG)

    @property
    def modellfehler(self) -> bool:
        """Liegt die Ursache im Modell (und nicht in der Rechenzeit)?"""
        return self in (SolverStatus.UNZULAESSIG, SolverStatus.UNBESCHRAENKT,
                        SolverStatus.FEHLERHAFT)


def status_von_pywraplp(rohstatus: int) -> SolverStatus:
    """OR-Tools linear_solver (GLOP, SCIP, CBC)."""
    from ortools.linear_solver import pywraplp
    zuordnung = {
        pywraplp.Solver.OPTIMAL: SolverStatus.OPTIMAL,
        pywraplp.Solver.FEASIBLE: SolverStatus.ZULAESSIG,
        pywraplp.Solver.INFEASIBLE: SolverStatus.UNZULAESSIG,
        pywraplp.Solver.UNBOUNDED: SolverStatus.UNBESCHRAENKT,
        pywraplp.Solver.ABNORMAL: SolverStatus.FEHLERHAFT,
        pywraplp.Solver.NOT_SOLVED: SolverStatus.ZEITLIMIT,
    }
    return zuordnung.get(rohstatus, SolverStatus.UNBEKANNT)


def status_von_cpsat(rohstatus: int) -> SolverStatus:
    """OR-Tools CP-SAT (siehe Kapitel CP-SAT, alle fuenf Faelle)."""
    from ortools.sat.python import cp_model
    zuordnung = {
        cp_model.OPTIMAL: SolverStatus.OPTIMAL,
        cp_model.FEASIBLE: SolverStatus.ZULAESSIG,
        cp_model.INFEASIBLE: SolverStatus.UNZULAESSIG,
        cp_model.MODEL_INVALID: SolverStatus.FEHLERHAFT,
        cp_model.UNKNOWN: SolverStatus.ZEITLIMIT,
    }
    return zuordnung.get(rohstatus, SolverStatus.UNBEKANNT)


def status_von_highs(statustext: str) -> SolverStatus:
    """HiGHS ueber highspy - hier kommt der Status als Text."""
    zuordnung = {
        "Optimal": SolverStatus.OPTIMAL,
        "Time limit reached": SolverStatus.ZULAESSIG,   # meist mit Loesung
        "Solution limit reached": SolverStatus.ZULAESSIG,
        "Infeasible": SolverStatus.UNZULAESSIG,
        "Unbounded": SolverStatus.UNBESCHRAENKT,
        "Primal infeasible or unbounded": SolverStatus.UNZULAESSIG,
        "Model error": SolverStatus.FEHLERHAFT,
    }
    return zuordnung.get(statustext, SolverStatus.UNBEKANNT)


def status_von_scipy(ergebnis: Any) -> SolverStatus:
    """scipy.optimize.linprog und milp liefern ein Ergebnisobjekt mit .status."""
    zuordnung = {
        0: SolverStatus.OPTIMAL,
        1: SolverStatus.ZEITLIMIT,        # Iterations-/Zeitgrenze
        2: SolverStatus.UNZULAESSIG,
        3: SolverStatus.UNBESCHRAENKT,
        4: SolverStatus.FEHLERHAFT,       # numerische Schwierigkeiten
    }
    return zuordnung.get(int(ergebnis.status), SolverStatus.UNBEKANNT)


def status_von_cvxpy(statustext: str) -> SolverStatus:
    """CVXPY - Statuszeichenketten wie 'optimal' oder 'infeasible'."""
    zuordnung = {
        "optimal": SolverStatus.OPTIMAL,
        "optimal_inaccurate": SolverStatus.ZULAESSIG,
        "infeasible": SolverStatus.UNZULAESSIG,
        "infeasible_inaccurate": SolverStatus.UNZULAESSIG,
        "unbounded": SolverStatus.UNBESCHRAENKT,
        "unbounded_inaccurate": SolverStatus.UNBESCHRAENKT,
    }
    return zuordnung.get(statustext, SolverStatus.UNBEKANNT)


# --- 2. Die Loesung als solverunabhaengiges Datenobjekt ----------------------

class Loesung(BaseModel):
    """Das Ergebnis eines Solverlaufs, so wie es weiterverarbeitet wird.

    Bewusst OHNE Referenz auf Solver, Modell oder Variablen: Ein Bericht, ein
    Test oder eine Weiterverarbeitung soll nicht wissen muessen, womit
    gerechnet wurde.
    """
    status: SolverStatus
    werte: dict[str, float] = Field(default_factory=dict)
    zielwert: float | None = None
    schranke: float | None = None
    laufzeit: float | None = None
    # Schattenpreise je Nebenbedingung, sofern der Solver sie liefert
    # (nur bei kontinuierlichen Problemen, siehe Kapitel LP). Auch das ist
    # ein solverunabhaengiger Begriff und gehoert deshalb hierher.
    schattenpreise: dict[str, float] = Field(default_factory=dict)

    @property
    def gap(self) -> float | None:
        """Relativer Abstand zwischen gefundener Loesung und bewiesener Schranke.

        Das ist die Zahl fuer den Bericht: keine Schaetzung, sondern eine
        Zusage - schlechter als das kann die Loesung nicht sein.
        """
        if self.zielwert is None or self.schranke is None:
            return None
        nenner = max(abs(self.zielwert), 1e-9)
        return abs(self.zielwert - self.schranke) / nenner

    def als_bericht(self) -> str:
        """Eine Zeile fuers Betriebsprotokoll - siehe Betriebsueberwachung.py."""
        teile = [f"Status: {self.status.value}"]
        if self.zielwert is not None:
            teile.append(f"Zielwert: {self.zielwert:,.2f}")
        if self.gap is not None:
            teile.append(f"Gap: {self.gap:.2%}")
        if self.laufzeit is not None:
            teile.append(f"Zeit: {self.laufzeit:.2f}s")
        return " | ".join(teile)


# --- 3. Domaenenmodell: die Fakten, bevor ein Solver sie sieht ---------------

class Produkt(BaseModel):
    """Ein Produkt mit Deckungsbeitrag und Ressourcenverbrauch."""
    name: str = Field(min_length=1)
    deckungsbeitrag: float
    verbrauch: dict[str, NichtNegativ]

    model_config = {"frozen": True}      # Stammdaten aendern sich nicht im Lauf


class Produktionsproblem(BaseModel):
    """Produktionsprogrammplanung - das Modell aus Kapitel Einfuehrung.

    Die Pruefungen stehen hier und nicht im Solvercode. Damit gelten sie
    unabhaengig davon, mit welcher Bibliothek spaeter gerechnet wird - und sie
    schlagen beim EINLESEN zu, wo der Fehler noch zuzuordnen ist.
    """
    produkte: list[Produkt] = Field(min_length=1)
    kapazitaeten: dict[str, PositiveZahl]

    @model_validator(mode="after")
    def pruefe_ressourcen(self) -> "Produktionsproblem":
        namen = [p.name for p in self.produkte]
        if len(set(namen)) != len(namen):
            doppelt = sorted({n for n in namen if namen.count(n) > 1})
            raise ValueError(f"Produktnamen kommen mehrfach vor: {doppelt}")

        for produkt in self.produkte:
            unbekannt = set(produkt.verbrauch) - set(self.kapazitaeten)
            if unbekannt:
                raise ValueError(
                    f"Produkt '{produkt.name}' verbraucht Ressourcen ohne "
                    f"Kapazitaetsangabe: {sorted(unbekannt)}")
        return self

    @property
    def ressourcen(self) -> list[str]:
        """Feste Reihenfolge - siehe die Spaltenfalle in Kapitel Finanzdaten."""
        return sorted(self.kapazitaeten)

    def verbrauchsmatrix(self) -> np.ndarray:
        """Zeilen = Ressourcen, Spalten = Produkte (Matrixform, Kapitel Fundament)."""
        return np.array([[p.verbrauch.get(r, 0.0) for p in self.produkte]
                         for r in self.ressourcen])

    def kapazitaetsvektor(self) -> np.ndarray:
        return np.array([self.kapazitaeten[r] for r in self.ressourcen])

    def deckungsbeitragsvektor(self) -> np.ndarray:
        return np.array([p.deckungsbeitrag for p in self.produkte])


# --- 4. Abnahmepruefung: Loesung gegen Anforderung, ohne Solver --------------

def pruefe_loesung(problem: Produktionsproblem, loesung: Loesung,
                   ganzzahlig: Sequence[str] = (),
                   toleranz: float = 1e-6) -> list[str]:
    """Prueft eine Loesung gegen die Anforderungen - OHNE den Solver zu fragen.

    Diese Funktion darf ausdruecklich KEIN Solverobjekt und keine
    Modellvariable benutzen. Eine Pruefung aus denselben Bausteinen wie das
    Modell prueft das Modell gegen sich selbst und findet nichts (siehe die
    Denkfehler in den Kapiteln Graphen und MILP).

    Liefert eine Liste von Beanstandungen; leer heisst bestanden.
    """
    beanstandungen: list[str] = []

    if not loesung.status.brauchbar:
        return [f"Kein verwertbares Ergebnis (Status: {loesung.status.value})"]

    fehlend = [p.name for p in problem.produkte if p.name not in loesung.werte]
    if fehlend:
        return [f"Loesung enthaelt keine Werte fuer: {fehlend}"]

    mengen = np.array([loesung.werte[p.name] for p in problem.produkte])

    if (mengen < -toleranz).any():
        negativ = [p.name for p, m in zip(problem.produkte, mengen) if m < -toleranz]
        beanstandungen.append(f"negative Mengen bei: {negativ}")

    verbrauch = problem.verbrauchsmatrix() @ mengen
    for ressource, ist, grenze in zip(problem.ressourcen, verbrauch,
                                      problem.kapazitaetsvektor()):
        if ist > grenze + toleranz:
            beanstandungen.append(
                f"{ressource}: Verbrauch {ist:,.3f} ueber Kapazitaet {grenze:,.3f}")

    for name in ganzzahlig:
        wert = loesung.werte[name]
        if abs(wert - round(wert)) > toleranz:
            beanstandungen.append(f"'{name}' = {wert!r} ist nicht ganzzahlig")

    if loesung.zielwert is not None:
        nachgerechnet = float(problem.deckungsbeitragsvektor() @ mengen)
        if abs(nachgerechnet - loesung.zielwert) > toleranz * max(1.0, abs(nachgerechnet)):
            beanstandungen.append(
                f"Zielwert {loesung.zielwert:,.4f} passt nicht zu den Mengen "
                f"(nachgerechnet {nachgerechnet:,.4f})")

    return beanstandungen


# --- 5. Ein Modellbauer je Solver - austauschbar ----------------------------

def loese_mit_glop(problem: Produktionsproblem) -> Loesung:
    """Modellbauer fuer OR-Tools GLOP (kontinuierlich)."""
    import time
    from ortools.linear_solver import pywraplp

    solver = pywraplp.Solver.CreateSolver("GLOP")
    menge = {p.name: solver.NumVar(0, solver.infinity(), p.name)
             for p in problem.produkte}
    # Die Nebenbedingung wird gemerkt - ohne die Referenz gibt es spaeter
    # keinen Schattenpreis (siehe Kapitel LP).
    bedingung = {}
    for ressource in problem.ressourcen:
        bedingung[ressource] = solver.Add(
            sum(menge[p.name] * p.verbrauch.get(ressource, 0.0)
                for p in problem.produkte)
            <= problem.kapazitaeten[ressource], name=ressource)
    solver.Maximize(sum(menge[p.name] * p.deckungsbeitrag for p in problem.produkte))

    t0 = time.perf_counter()
    rohstatus = solver.Solve()
    laufzeit = time.perf_counter() - t0

    status = status_von_pywraplp(rohstatus)
    if not status.brauchbar:
        return Loesung(status=status, laufzeit=laufzeit)
    return Loesung(status=status,
                   werte={name: v.solution_value() for name, v in menge.items()},
                   zielwert=solver.Objective().Value(),
                   schranke=solver.Objective().Value(),
                   schattenpreise={r: b.dual_value() for r, b in bedingung.items()},
                   laufzeit=laufzeit)


def loese_mit_scipy(problem: Produktionsproblem) -> Loesung:
    """Derselbe Fall mit scipy.optimize.linprog - anderer Solver, gleiche Loesung.

    Beachten Sie, wie wenig hier steht: Die Daten kommen fertig geprueft aus
    dem Domaenenmodell, die Auswertung geht ans gemeinsame Loesung-Objekt.
    Genau das ist der Ertrag der Trennung.
    """
    import time
    from scipy.optimize import linprog

    t0 = time.perf_counter()
    ergebnis = linprog(c=-problem.deckungsbeitragsvektor(),      # linprog minimiert
                       A_ub=problem.verbrauchsmatrix(),
                       b_ub=problem.kapazitaetsvektor(),
                       bounds=(0, None), method="highs")
    laufzeit = time.perf_counter() - t0

    status = status_von_scipy(ergebnis)
    if not status.brauchbar:
        return Loesung(status=status, laufzeit=laufzeit)
    # linprog rechnet minimierend mit negierten Kosten - die Dualwerte
    # muessen entsprechend zurueckgedreht werden.
    return Loesung(status=status,
                   werte={p.name: float(x)
                          for p, x in zip(problem.produkte, ergebnis.x)},
                   zielwert=float(-ergebnis.fun),
                   schranke=float(-ergebnis.fun),
                   schattenpreise={r: float(-m) for r, m in
                                   zip(problem.ressourcen,
                                       ergebnis.ineqlin.marginals)},
                   laufzeit=laufzeit)


# --- 6. Ein- und Ausgabe: Tabellen sind die Wirklichkeit ---------------------
#
# In den meisten Betrieben liegen die Zahlen in einer Tabellenkalkulation
# (Kapitel Einfuehrung). Diese beiden Funktionen sind die einzige Stelle im
# Modul, die das weiss - alles andere arbeitet mit dem Domaenenmodell.
#
# pandas und openpyxl werden bewusst erst hier importiert: Wer or_kern nur
# fuer Statusauswertung und Pruefung benutzt, soll sie nicht installieren
# muessen.

STAMMSPALTEN = ("Produkt", "Deckungsbeitrag")


def lade_produktionsproblem(pfad: str, blatt_produkte: str = "Produkte",
                            blatt_kapazitaeten: str = "Kapazitaeten"
                            ) -> Produktionsproblem:
    """Liest eine Excel-Mappe und liefert ein geprueftes Domaenenmodell.

    Alle Spalten ausser 'Produkt' und 'Deckungsbeitrag' gelten als
    Verbrauchsspalten - eine neue Ressource ist damit eine neue SPALTE in der
    Mappe und erfordert keine Codeaenderung.

    Die eigentliche Pruefung passiert nicht hier, sondern im Konstruktor von
    Produktionsproblem. Diese Funktion uebersetzt nur Tabelle in Objekt; was
    ein gueltiges Problem ist, steht an genau einer Stelle.
    """
    import pandas as pd

    produkte_tabelle = pd.read_excel(pfad, sheet_name=blatt_produkte)
    kapazitaeten_tabelle = pd.read_excel(pfad, sheet_name=blatt_kapazitaeten)

    fehlende = [s for s in STAMMSPALTEN if s not in produkte_tabelle.columns]
    if fehlende:
        raise ValueError(f"Blatt '{blatt_produkte}': Spalten fehlen: {fehlende}")
    if produkte_tabelle.isna().any().any():
        zeilen = produkte_tabelle[produkte_tabelle.isna().any(axis=1)].index.tolist()
        raise ValueError(f"Blatt '{blatt_produkte}': leere Zellen in Zeile(n) {zeilen}")

    ressourcen = [s for s in produkte_tabelle.columns if s not in STAMMSPALTEN]
    produkte = [
        Produkt(name=str(zeile.Produkt),
                deckungsbeitrag=float(zeile.Deckungsbeitrag),
                verbrauch={r: float(getattr(zeile, r)) for r in ressourcen})
        for zeile in produkte_tabelle.itertuples()
    ]
    kapazitaeten = {str(z.Ressource): float(z.Verfuegbar)
                    for z in kapazitaeten_tabelle.itertuples()}

    return Produktionsproblem(produkte=produkte, kapazitaeten=kapazitaeten)


def schreibe_ergebnis(pfad: str, problem: Produktionsproblem,
                      loesung: Loesung) -> None:
    """Schreibt Plan und Kennzahlen als zwei Blaetter zurueck nach Excel.

    Die Fachabteilung bekommt das Ergebnis im Format, das sie ohnehin benutzt.
    Akzeptanz entscheidet ueber Projekterfolg (Kapitel Praxisfallen).
    """
    import pandas as pd

    if not loesung.status.brauchbar:
        raise ValueError(f"Nichts zu schreiben - Status: {loesung.status.value}")

    mengen = np.array([loesung.werte[p.name] for p in problem.produkte])
    plan = pd.DataFrame({
        "Produkt": [p.name for p in problem.produkte],
        "Menge": mengen,
        "Deckungsbeitrag": mengen * problem.deckungsbeitragsvektor(),
    })

    verbrauch = problem.verbrauchsmatrix() @ mengen
    kennzahlen: dict[str, float | str] = {
        "Status": loesung.status.value,
        "Gesamtdeckungsbeitrag": loesung.zielwert or 0.0,
    }
    for ressource, ist, grenze in zip(problem.ressourcen, verbrauch,
                                      problem.kapazitaetsvektor()):
        kennzahlen[f"{ressource}: Verbrauch"] = float(ist)
        kennzahlen[f"{ressource}: Auslastung %"] = float(100.0 * ist / grenze)
        if ressource in loesung.schattenpreise:
            kennzahlen[f"{ressource}: Schattenpreis"] = \
                loesung.schattenpreise[ressource]

    kennzahlen_tabelle = pd.DataFrame({
        "Kennzahl": list(kennzahlen),
        "Wert": [round(w, 2) if isinstance(w, float) else w
                 for w in kennzahlen.values()],
    })

    with pd.ExcelWriter(pfad, engine="openpyxl") as mappe:
        plan.round(2).to_excel(mappe, sheet_name="Plan", index=False)
        kennzahlen_tabelle.to_excel(mappe, sheet_name="Kennzahlen", index=False)


if __name__ == "__main__":
    # Die Schreinerei aus Kapitel Einfuehrung - jetzt als Domaenenmodell.
    schreinerei = Produktionsproblem(
        produkte=[
            Produkt(name="Tisch", deckungsbeitrag=240.0,
                    verbrauch={"Montagestunden": 3.0, "Plattenmaterial": 6.0}),
            Produkt(name="Stuhl", deckungsbeitrag=60.0,
                    verbrauch={"Montagestunden": 1.0, "Plattenmaterial": 1.0}),
        ],
        kapazitaeten={"Montagestunden": 150.0, "Plattenmaterial": 240.0},
    )

    print("=" * 78)
    print("  DERSELBE FALL, ZWEI SOLVER, EIN AUSWERTUNGSCODE")
    print("=" * 78)

    for name, bauer in [("OR-Tools GLOP", loese_mit_glop),
                        ("scipy / HiGHS", loese_mit_scipy)]:
        loesung = bauer(schreinerei)
        beanstandungen = pruefe_loesung(schreinerei, loesung)
        print(f"\n{name}")
        print(f"  {loesung.als_bericht()}")
        for produkt, menge in loesung.werte.items():
            print(f"    {produkt:<10} {menge:8.2f}")
        print(f"  Abnahmepruefung: "
              f"{'bestanden' if not beanstandungen else beanstandungen}")

    print("\n" + "=" * 78)
    print("  WAS DIE PRUEFUNGEN ABFANGEN")
    print("=" * 78)

    faelle = [
        ("Kapazitaet 0", dict(
            produkte=schreinerei.produkte,
            kapazitaeten={"Montagestunden": 0.0, "Plattenmaterial": 240.0})),
        ("Ressource ohne Kapazitaet", dict(
            produkte=[Produkt(name="Regal", deckungsbeitrag=130.0,
                              verbrauch={"Lackieren": 2.0})],
            kapazitaeten={"Montagestunden": 150.0})),
        ("doppelter Produktname", dict(
            produkte=[schreinerei.produkte[0], schreinerei.produkte[0]],
            kapazitaeten=schreinerei.kapazitaeten)),
    ]
    for beschreibung, daten in faelle:
        try:
            Produktionsproblem(**daten)
            print(f"  {beschreibung:<28} NICHT erkannt (!)")
        except Exception as fehler:
            meldung = str(fehler).splitlines()
            kern = next((z.strip() for z in meldung
                         if "Value error" in z or "greater than" in z), meldung[-1])
            print(f"  {beschreibung:<28} abgefangen: {kern[:44]}")

    print("\nAlle drei scheitern beim EINLESEN - nicht erst beim Loesen und")
    print("schon gar nicht erst im Bericht. Das ist der ganze Zweck der")
    print("Domaenenschicht.")
    print("=" * 78)

#!/usr/bin/env python3

# Solverwechsel_CPSAT_HiGHS.py
"""
Kapitel Praxisfallen: Denselben Fall einmal mit CP-SAT und einmal mit HiGHS rechnen.

Die Trennung aus or_kern.py behauptet, ein Solverwechsel koste genau EINEN
Baustein. Dieses Programm loest diese Behauptung ein. Aufgabe ist eine
Standortplanung: Welche Lager oeffnen wir, und wer beliefert welchen Kunden?

    Standortproblem  (Pydantic, geprueft)   <- gemeinsam
        baue_und_loese_mit_cpsat()          <- der EINE Baustein
        baue_und_loese_mit_highs()          <- ... in zwei Ausfuehrungen
    Loesung          (DTO)                  <- gemeinsam
    pruefe_zuordnung()                      <- gemeinsam
    berichte()                              <- gemeinsam

Die beiden Modellbauer sind 39 und 52 Zeilen lang - sie sind der gesamte
solverabhaengige Teil des Programms. Alles andere wird zweimal benutzt und
einmal geschrieben.

WARUM ZWEI PROZESSE? ortools und highspy bringen beide eine eigene
HiGHS-Kopie mit und lassen sich auf vielen Systemen nicht gemeinsam
importieren (Kapitel Oekosystem). Das Hauptprogramm startet deshalb fuer jeden
Solver einen eigenen Python-Prozess und laesst sich die Loesung als JSON
zurueckgeben - das DTO ist nicht nur eine Sprachregelung, sondern ein
Datenformat, das eine Prozessgrenze ueberlebt.

Aufruf:
    python3 Solverwechsel_CPSAT_HiGHS.py            # beide, mit Vergleich
    python3 Solverwechsel_CPSAT_HiGHS.py cpsat      # nur der Kindprozess
    python3 Solverwechsel_CPSAT_HiGHS.py highs

Benoetigt: numpy, pydantic, ortools, highspy (jeweils im eigenen Prozess)
"""

from __future__ import annotations

import subprocess
import sys
import time

import numpy as np
from pydantic import BaseModel, Field, model_validator

from or_kern import Loesung, SolverStatus, status_von_cpsat, status_von_highs

ZEITLIMIT = 30.0


# --- 1. Domaenenmodell: dieselben Fakten fuer beide Solver -------------------

class Standortproblem(BaseModel):
    """Kapazitierte Standortplanung mit Einzelbelieferung.

    Nach dem Muster von Produktionsproblem in or_kern.py: Die Pruefungen
    stehen im Konstruktor, nicht im Solvercode - sie gelten damit fuer beide
    Solver, und sie schlagen beim Einlesen zu.

    Alle Kosten sind ganzzahlig (Euro). Das ist keine Bequemlichkeit, sondern
    Voraussetzung: CP-SAT rechnet ausschliesslich ganzzahlig.
    """
    lager: list[str] = Field(min_length=1)
    kunden: list[str] = Field(min_length=1)
    fixkosten: list[int]        # je Lager, faellt bei Eroeffnung an
    kapazitaet: list[int]       # je Lager, in Paletten
    bedarf: list[int]           # je Kunde, in Paletten
    transport: list[list[int]]  # [Lager][Kunde], Kosten der Belieferung

    @model_validator(mode="after")
    def pruefe_masse(self) -> "Standortproblem":
        n, m = len(self.lager), len(self.kunden)
        if len(self.fixkosten) != n or len(self.kapazitaet) != n:
            raise ValueError(f"fixkosten/kapazitaet muessen {n} Eintraege haben")
        if len(self.bedarf) != m:
            raise ValueError(f"bedarf muss {m} Eintraege haben")
        if len(self.transport) != n or any(len(z) != m for z in self.transport):
            raise ValueError(f"transport muss {n} x {m} sein")
        if sum(self.kapazitaet) < sum(self.bedarf):
            raise ValueError(f"Gesamtkapazitaet {sum(self.kapazitaet)} deckt den "
                             f"Gesamtbedarf {sum(self.bedarf)} nicht")
        return self

    def schluessel(self, i: int, j: int) -> str:
        """Variablenname im Loesung-DTO - beide Modellbauer benutzen ihn."""
        return f"{self.lager[i]}->{self.kunden[j]}"


def beispielproblem(saat: int = 11) -> Standortproblem:
    """Sechs moegliche Lager, zwoelf Kunden - klein genug fuer beide Solver."""
    rng = np.random.default_rng(saat)
    lager = [f"Lager_{k}" for k in "ABCDEF"]
    kunden = [f"Kunde_{k:02d}" for k in range(1, 13)]
    bedarf = rng.integers(10, 60, len(kunden))
    return Standortproblem(
        lager=lager,
        kunden=kunden,
        fixkosten=rng.integers(3000, 9000, len(lager)).tolist(),
        kapazitaet=[int(bedarf.sum() * 0.45)] * len(lager),
        bedarf=bedarf.tolist(),
        transport=rng.integers(200, 1800, (len(lager), len(kunden))).tolist(),
    )


# --- 2. Der eine Baustein, der sich aendert: der Modellbauer -----------------

def baue_und_loese_mit_cpsat(problem: Standortproblem) -> Loesung:
    """CP-SAT: Bool-Variablen, ganzzahlige Koeffizienten, Minimize."""
    from ortools.sat.python import cp_model

    n, m = len(problem.lager), len(problem.kunden)
    modell = cp_model.CpModel()
    y = [modell.NewBoolVar(f"offen_{i}") for i in range(n)]
    x = {(i, j): modell.NewBoolVar(f"liefert_{i}_{j}")
         for i in range(n) for j in range(m)}

    for j in range(m):                                   # jeder Kunde genau einmal
        modell.AddExactlyOne(x[i, j] for i in range(n))
    for i in range(n):
        for j in range(m):                               # nur aus offenen Lagern
            modell.AddImplication(x[i, j], y[i])
        modell.Add(sum(problem.bedarf[j] * x[i, j] for j in range(m))
                   <= problem.kapazitaet[i] * y[i])      # Kapazitaet

    modell.Minimize(
        sum(problem.fixkosten[i] * y[i] for i in range(n))
        + sum(problem.transport[i][j] * x[i, j] for i in range(n) for j in range(m)))

    loeser = cp_model.CpSolver()
    loeser.parameters.max_time_in_seconds = ZEITLIMIT
    loeser.parameters.num_workers = 1
    loeser.parameters.random_seed = 1
    status = status_von_cpsat(loeser.Solve(modell))
    if not status.brauchbar:
        return Loesung(status=status, laufzeit=loeser.WallTime())

    return Loesung(
        status=status,
        werte={problem.schluessel(i, j): float(loeser.Value(x[i, j]))
               for i in range(n) for j in range(m)},
        zielwert=loeser.ObjectiveValue(),
        schranke=loeser.BestObjectiveBound(),
        laufzeit=loeser.WallTime())


def baue_und_loese_mit_highs(problem: Standortproblem) -> Loesung:
    """HiGHS: dieselben Restriktionen als Ungleichungszeilen einer Matrix."""
    import highspy

    n, m = len(problem.lager), len(problem.kunden)
    anzahl_x = n * m                                     # Spalten 0..n*m-1
    spalte_y = lambda i: anzahl_x + i                    # danach die y_i  # noqa: E731

    modell = highspy.Highs()
    modell.setOptionValue("output_flag", False)
    modell.setOptionValue("time_limit", ZEITLIMIT)
    modell.addVars(anzahl_x + n, np.zeros(anzahl_x + n), np.ones(anzahl_x + n))
    for spalte in range(anzahl_x + n):
        modell.changeColIntegrality(spalte, highspy.HighsVarType.kInteger)
    for i in range(n):
        modell.changeColCost(spalte_y(i), float(problem.fixkosten[i]))
        for j in range(m):
            modell.changeColCost(i * m + j, float(problem.transport[i][j]))

    for j in range(m):                                   # jeder Kunde genau einmal
        index = np.array([i * m + j for i in range(n)], dtype=np.int32)
        modell.addRow(1.0, 1.0, n, index, np.ones(n))
    for i in range(n):
        for j in range(m):                               # x_ij - y_i <= 0
            modell.addRow(-highspy.kHighsInf, 0.0, 2,
                          np.array([i * m + j, spalte_y(i)], dtype=np.int32),
                          np.array([1.0, -1.0]))
        index = np.array([i * m + j for j in range(m)] + [spalte_y(i)],
                         dtype=np.int32)                 # Kapazitaet
        werte = np.concatenate([np.array(problem.bedarf, dtype=float),
                                [-float(problem.kapazitaet[i])]])
        modell.addRow(-highspy.kHighsInf, 0.0, m + 1, index, werte)

    t0 = time.perf_counter()
    modell.run()
    laufzeit = time.perf_counter() - t0

    status = status_von_highs(modell.modelStatusToString(modell.getModelStatus()))
    if not status.brauchbar:
        return Loesung(status=status, laufzeit=laufzeit)

    loesungswerte = modell.getSolution().col_value
    info = modell.getInfo()
    return Loesung(
        status=status,
        werte={problem.schluessel(i, j): float(loesungswerte[i * m + j])
               for i in range(n) for j in range(m)},
        zielwert=info.objective_function_value,
        schranke=info.mip_dual_bound,
        laufzeit=laufzeit)


MODELLBAUER = {"cpsat": baue_und_loese_mit_cpsat, "highs": baue_und_loese_mit_highs}


# --- 3. Alles Weitere ist wieder gemeinsam -----------------------------------

def pruefe_zuordnung(problem: Standortproblem, loesung: Loesung,
                     toleranz: float = 1e-6) -> list[str]:
    """Prueft die Loesung gegen die Anforderung - ohne Solver, ohne Modell."""
    if not loesung.status.brauchbar:
        return [f"kein verwertbares Ergebnis ({loesung.status.value})"]

    n, m = len(problem.lager), len(problem.kunden)
    zuordnung = np.array([[loesung.werte[problem.schluessel(i, j)]
                           for j in range(m)] for i in range(n)])
    beanstandungen: list[str] = []

    if (np.abs(zuordnung - np.round(zuordnung)) > toleranz).any():
        beanstandungen.append("Zuordnungen sind nicht 0/1")
    zuordnung = np.round(zuordnung)

    for j, kunde in enumerate(problem.kunden):
        if abs(zuordnung[:, j].sum() - 1.0) > toleranz:
            beanstandungen.append(f"{kunde} wird {zuordnung[:, j].sum():.0f}-mal beliefert")

    beliefert = zuordnung @ np.array(problem.bedarf, dtype=float)
    for i, lagername in enumerate(problem.lager):
        if beliefert[i] > problem.kapazitaet[i] + toleranz:
            beanstandungen.append(f"{lagername}: {beliefert[i]:.0f} Paletten ueber "
                                  f"Kapazitaet {problem.kapazitaet[i]}")

    if loesung.zielwert is not None:
        offen = beliefert > toleranz
        nachgerechnet = (np.array(problem.fixkosten, dtype=float) @ offen
                         + (np.array(problem.transport, dtype=float) * zuordnung).sum())
        if abs(nachgerechnet - loesung.zielwert) > 0.5:
            beanstandungen.append(f"Zielwert {loesung.zielwert:,.0f} passt nicht zur "
                                  f"Zuordnung (nachgerechnet {nachgerechnet:,.0f})")
    return beanstandungen


def geoeffnete_lager(problem: Standortproblem, loesung: Loesung) -> list[str]:
    m = len(problem.kunden)
    return [name for i, name in enumerate(problem.lager)
            if any(loesung.werte[problem.schluessel(i, j)] > 0.5 for j in range(m))]


def loese_in_eigenem_prozess(name: str) -> Loesung:
    """Startet dieses Programm noch einmal - mit genau einem Solverimport."""
    ergebnis = subprocess.run([sys.executable, __file__, name],
                              capture_output=True, text=True, timeout=300)
    if ergebnis.returncode != 0:
        raise RuntimeError(ergebnis.stderr.strip().splitlines()[-1])
    # Das DTO als JSON - genau dafuer ist ein Datenobjekt ohne Solverbezug gut.
    return Loesung.model_validate_json(ergebnis.stdout.strip().splitlines()[-1])


if __name__ == "__main__":
    problem = beispielproblem()

    # --- Kindprozess: rechnen und das DTO als JSON ausgeben ---------------
    if len(sys.argv) > 1:
        print(MODELLBAUER[sys.argv[1]](problem).model_dump_json())
        sys.exit(0)

    # --- Hauptprozess: beide Solver anstossen und vergleichen -------------
    print("=" * 82)
    print("  DERSELBE FALL, ZWEI SOLVER - UND EIN AUSWERTUNGSCODE")
    print("=" * 82)
    print(f"Standortplanung: {len(problem.lager)} moegliche Lager, "
          f"{len(problem.kunden)} Kunden, {sum(problem.bedarf)} Paletten Bedarf.")
    print(f"Kapazitaet je Lager: {problem.kapazitaet[0]} Paletten "
          f"-> mindestens 3 Lager noetig.\n")

    loesungen: dict[str, Loesung] = {}
    for name, beschriftung in [("cpsat", "OR-Tools CP-SAT"),
                               ("highs", "HiGHS (highspy)")]:
        loesung = loesungen[name] = loese_in_eigenem_prozess(name)
        beanstandungen = pruefe_zuordnung(problem, loesung)

        print(f"{beschriftung}")
        print(f"  {loesung.als_bericht()}")
        print(f"  eroeffnete Lager: {', '.join(geoeffnete_lager(problem, loesung))}")
        print(f"  Abnahmepruefung:  "
              f"{'bestanden' if not beanstandungen else beanstandungen}")

    # --- Was der Vergleich zeigt -----------------------------------------
    zielwerte = [loesung.zielwert for loesung in loesungen.values()]
    print("-" * 82)
    print(f"Zielwertdifferenz: {abs(zielwerte[0] - zielwerte[1]):.6f} EUR")

    gleich_belegt = all(
        round(loesungen["cpsat"].werte[s]) == round(loesungen["highs"].werte[s])
        for s in loesungen["cpsat"].werte)
    print(f"Identische Zuordnung: {'ja' if gleich_belegt else 'nein'}")

    assert abs(zielwerte[0] - zielwerte[1]) < 0.5, "Die Solver widersprechen sich!"
    assert all(l.status is SolverStatus.OPTIMAL for l in loesungen.values())

    print("\n" + "=" * 82)
    print("  WAS DER WECHSEL GEKOSTET HAT")
    print("=" * 82)
    print("Ausgetauscht wurde EINE Funktion. Domaenenmodell, Abnahmepruefung und")
    print("Bericht sind woertlich dieselben - sie sehen den Solver nie.")
    print()
    print("Nicht umsonst ist der Wechsel trotzdem:")
    print("  * CP-SAT rechnet ausschliesslich GANZZAHLIG. Alle Kosten sind hier")
    print("    deshalb int. Wer in Euro und Cent rechnet, skaliert vorher auf Cent -")
    print("    und muss das im Bericht wieder zuruecknehmen.")
    print("  * HiGHS braucht die Restriktionen als Matrixzeilen, CP-SAT nimmt sie")
    print("    als Ausdruecke. Das ist der Grund, warum der HiGHS-Modellbauer")
    print("    laenger ist, obwohl er dasselbe Modell beschreibt.")
    print("  * Beide Bibliotheken bringen eine eigene HiGHS-Kopie mit und lassen")
    print("    sich nicht gemeinsam importieren - daher die zwei Prozesse.")
    print()
    print("Der Ertrag: Beide beweisen denselben optimalen Zielwert, und die")
    print("Entscheidung zwischen ihnen ist eine Frage der Laufzeit geworden -")
    print("nicht eine Frage, wie viel Code man neu schreiben muss.")
    print()
    print("Verglichen wird deshalb der ZIELWERT, nicht der Plan: Gibt es mehrere")
    print("gleich teure Loesungen, darf jeder Solver eine andere davon liefern.")
    print("Hier stimmen sie zufaellig ueberein - darauf zu testen waere trotzdem")
    print("ein unzuverlaessiger Test (siehe JobShop_Intervalle.py).")
    print("=" * 82)

#!/usr/bin/env python3

# Betriebsueberwachung.py
"""
Kapitel Praxisfallen: Das Modell, das still aufgehoert hat zu optimieren.

Ein Optimierungsmodell geht in Betrieb und laeuft jede Nacht. Es liefert
zuverlaessig eine Zahl, die Job-Dauer bleibt konstant, es gibt keine
Fehlermeldung. Alles sieht gut aus.

Waehrenddessen waechst das Unternehmen. Aus 15 Lagern werden 40, dann 80. Das
Zeitlimit von drei Sekunden - vor zwei Jahren grosszuegig bemessen - reicht
irgendwann nicht mehr. Der Solver bricht ab und liefert die beste Loesung, die
er bis dahin gefunden hat.

Das ist nicht falsch. Es ist sogar richtig so. Aber es ist etwas anderes als
das, was das Modell einmal geleistet hat - und niemand merkt es, weil im
Protokoll nur die Kosten stehen.

Dieses Programm zeigt die Verschlechterung und die Ueberwachung, die sie
sichtbar macht.

WICHTIG: ortools wird hier nicht importiert (Konflikt mit highspy, siehe
Kapitel Oekosystem).

Benoetigt: numpy, highspy
"""

from __future__ import annotations

import time
from dataclasses import dataclass

import numpy as np
import highspy

ZEITLIMIT = 3.0                  # das Budget, das der Nachtjob hat
GAP_WARNSCHWELLE = 0.02          # ab 2 % Gap wollen wir es wissen
ZEITAUSLASTUNG_WARNSCHWELLE = 0.9


@dataclass
class Laufprotokoll:
    """Was nach JEDEM Produktivlauf protokolliert gehoert.

    Die meisten Systeme schreiben nur 'kosten'. Genau deshalb faellt eine
    schleichende Verschlechterung ueber Monate nicht auf.
    """
    zeitpunkt: str
    lager: int
    kunden: int
    status: str
    kosten: float
    gap: float
    laufzeit: float
    knoten: int

    @property
    def warnungen(self) -> list[str]:
        meldungen = []
        if self.status != "Optimal":
            meldungen.append(f"nicht beweisbar optimal ({self.status})")
        if self.gap > GAP_WARNSCHWELLE:
            meldungen.append(f"Gap {self.gap:.1%} ueber Schwelle "
                             f"{GAP_WARNSCHWELLE:.0%}")
        if self.laufzeit > ZEITAUSLASTUNG_WARNSCHWELLE * ZEITLIMIT:
            meldungen.append(f"Zeitbudget zu {self.laufzeit / ZEITLIMIT:.0%} "
                             f"ausgeschoepft")
        return meldungen


def plane_netzwerk(n_lager: int, n_kunden: int, zeitlimit: float,
                   saat: int = 7) -> tuple[str, float, float, float, int]:
    """Standortplanung wie in Kapitel MILP, nur mit wachsender Groesse."""
    rng = np.random.default_rng(saat)
    fixkosten = rng.uniform(3000, 9000, n_lager)
    transport = rng.uniform(5, 60, (n_lager, n_kunden))
    bedarf = rng.uniform(10, 60, n_kunden)
    kapazitaet = np.full(n_lager, bedarf.sum() * 0.22)

    modell = highspy.Highs()
    modell.setOptionValue("output_flag", False)
    modell.setOptionValue("time_limit", zeitlimit)

    anzahl_x = n_lager * n_kunden
    unendlich = highspy.kHighsInf
    modell.addVars(anzahl_x, np.zeros(anzahl_x), np.full(anzahl_x, unendlich))
    modell.addVars(n_lager, np.zeros(n_lager), np.ones(n_lager))
    for i in range(n_lager):
        modell.changeColIntegrality(anzahl_x + i, highspy.HighsVarType.kInteger)
        modell.changeColCost(anzahl_x + i, fixkosten[i])
        for j in range(n_kunden):
            modell.changeColCost(i * n_kunden + j, transport[i, j])

    for j in range(n_kunden):
        index = np.array([i * n_kunden + j for i in range(n_lager)], dtype=np.int32)
        modell.addRow(bedarf[j], bedarf[j], len(index), index, np.ones(len(index)))
    for i in range(n_lager):
        index = np.array([i * n_kunden + j for j in range(n_kunden)] + [anzahl_x + i],
                         dtype=np.int32)
        werte = np.concatenate([np.ones(n_kunden), [-kapazitaet[i]]])
        modell.addRow(-unendlich, 0.0, len(index), index, werte)

    t0 = time.perf_counter()
    modell.run()
    laufzeit = time.perf_counter() - t0
    info = modell.getInfo()
    return (modell.modelStatusToString(modell.getModelStatus()),
            info.objective_function_value, info.mip_gap, laufzeit,
            info.mip_node_count)


if __name__ == "__main__":
    # So ist das Unternehmen ueber zwei Jahre gewachsen.
    entwicklung = [("2024-Q1", 15, 40), ("2024-Q3", 25, 70),
                   ("2025-Q1", 40, 110), ("2025-Q3", 60, 160),
                   ("2026-Q1", 80, 220)]

    print("=" * 84)
    print("  WAS DAS PROTOKOLL ZEIGT - UND WAS ES ZEIGEN SOLLTE")
    print("=" * 84)
    print(f"Derselbe Nachtjob, unveraendert, mit {ZEITLIMIT:.0f} Sekunden Zeitlimit.\n")

    protokolle = []
    for zeitpunkt, lager, kunden in entwicklung:
        status, kosten, gap, laufzeit, knoten = plane_netzwerk(
            lager, kunden, ZEITLIMIT)
        protokolle.append(Laufprotokoll(zeitpunkt, lager, kunden, status,
                                        kosten, gap, laufzeit, knoten))

    print("So sieht das ueblich gefuehrte Protokoll aus:\n")
    print(f"  {'Zeitpunkt':<10} {'Kosten':>12}")
    print("  " + "-" * 24)
    for p in protokolle:
        print(f"  {p.zeitpunkt:<10} {p.kosten:>12,.0f}")
    print("\n  -> Die Kosten steigen. Das Unternehmen waechst ja auch.")
    print("     Nichts an dieser Tabelle deutet auf ein Problem hin.")

    print("\n" + "-" * 84)
    print("So sieht ein vollstaendiges Protokoll aus:\n")
    print(f"  {'Zeitpunkt':<10} {'Groesse':>10} {'Status':>20} {'Kosten':>11} "
          f"{'Gap':>8} {'Zeit':>7}")
    print("  " + "-" * 80)
    for p in protokolle:
        groesse = f"{p.lager}x{p.kunden}"
        print(f"  {p.zeitpunkt:<10} {groesse:>10} {p.status:>20} "
              f"{p.kosten:>11,.0f} {p.gap * 100:7.2f} % {p.laufzeit:6.2f}s")

    print("\n" + "-" * 84)
    print("Und so sehen die Warnungen aus, die daraus folgen:\n")
    for p in protokolle:
        if p.warnungen:
            print(f"  {p.zeitpunkt}:  " + "; ".join(p.warnungen))
        else:
            print(f"  {p.zeitpunkt}:  in Ordnung")

    letzte = protokolle[-1]
    erste_warnung = next(p for p in protokolle if p.warnungen)

    print("\n" + "=" * 84)
    print("  WAS DA PASSIERT IST")
    print("=" * 84)
    print(f"Seit {erste_warnung.zeitpunkt} erreicht der Job das Zeitlimit und liefert")
    print("nicht mehr das Optimum, sondern die beste bis dahin gefundene Loesung.")
    print(f"Im letzten Lauf betraegt der Abstand zum Bestmoeglichen "
          f"{letzte.gap:.1%}.")
    print()
    print("Der Job ist nicht abgestuerzt. Er hat keine Fehlermeldung erzeugt.")
    print("Die Laufzeit ist sogar bemerkenswert STABIL geblieben - genau deshalb,")
    print("weil das Zeitlimit greift. Ein Ueberwachungssystem, das auf Abstuerze")
    print("und Laufzeitspitzen achtet, sieht hier nichts.")
    print()
    print("Drei Zahlen gehoeren deshalb in jedes Protokoll eines Optimierungsjobs:")
    print("  * der STATUS   - 'Optimal' oder etwas anderes?")
    print("  * der GAP      - wie weit ist die Loesung vom Bestmoeglichen entfernt?")
    print("  * die LAUFZEIT im Verhaeltnis zum Limit - wie nah am Anschlag?")
    print()
    print("Die dritte ist die frueheste Warnung: Sie steigt, lange bevor der Gap")
    print("sichtbar wird, und gibt Zeit zum Handeln.")
    print("=" * 84)