862e92bc7b
Setzt den Isolationsteil von Paket 1 aus Verbesserungen_02.md um. Der Plan
nannte zwei Programme; beim Suchen kam ein drittes dazu, das dasselbe Muster
verwendete.
Ein_System_Vier_Ansaetze.py und Benchmark_Skalierung.py hielten ihre vier
Solvervarianten als Zeichenketten in einem Dictionary und gaben sie an
"python -c" weiter - bei Benchmark_Skalierung.py sogar mit
.format()-Platzhaltern fuer die Instanzgroesse. Aus jeder Variante ist jetzt
eine gewoehnliche Funktion mit lokalem Import geworden.
Solverwechsel_CPSAT_HiGHS.py rief sich selbst ueber sys.argv erneut auf;
auch das entfaellt.
Ausgefuehrt wird ueber einen ProcessPoolExecutor mit zwei Einstellungen, die
beide noetig sind: mp_context "spawn" (frischer Interpreter statt geerbtem
Speicher - unter Linux ist fork der Standard) und max_tasks_per_child=1 (ein
neuer Prozess je Aufgabe; ohne das verwendet der Pool seinen Arbeiter
wieder, und beim zweiten Solver ist der Konflikt zurueck). Nachgemessen:
vier Aufgaben, vier verschiedene PIDs.
Der zweite Punkt hat einen eigenen Warnkasten bekommen, weil der Fehler
leicht zu machen und schwer zu finden ist: Der Absturz kaeme nicht beim
ersten Solver, sondern beim zweiten - und saehe aus wie ein Problem des
zweiten.
Regel 4, dreifach geprueft. Ein_System_Vier_Ansaetze.py: identisch bis auf
die Zeitspalte, einschliesslich der Spannweite 2,41e-08, auf die sich der
Merksatz des Kapitels beruft. Benchmark_Skalierung.py: alle zwoelf
Zielwerte und alle drei Spannweiten bitgleich; Zeiten und Speicher haben
sich verschoben, beide sind im Abdruck seit jeher als hardwareabhaengig
gekennzeichnet. Solverwechsel_CPSAT_HiGHS.py: Ausgabe ohne Zeiten
unveraendert.
Bewusst subprocess bleibt Mutationstest.py: Dort wird pytest auf einer
mutierten Kopie in einem temporaeren Verzeichnis gestartet - ein externes
Werkzeug auf veraenderten Dateien, nicht die Isolation eines Imports.
Neu im Kapitel Oekosystem: ein Abschnitt "Wie die Isolation aussieht, wenn
sie tragen soll" - warum ein Codestring die schlechteste Umsetzung von
"eigener Prozess" ist. Anhang C nennt jetzt ebenfalls ProcessPoolExecutor.
Ein eigener Fehler, gefunden und abgesichert: Ich hatte dem neuen ### ein
{#sec:...}-Label gegeben. ABSCHNITT_RE erkennt nur "## " - das Label waere
nie registriert worden und jeder Verweis darauf ins Leere gelaufen, ohne
Warnung. Label entfernt, --check meldet den Fall jetzt. Gegengetestet.
Stand: 818 Querverweise, 76 Programme, 33 pytest-Tests, PDF 760 Seiten, 69
netzfreie Programme fehlerfrei.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
1650 lines
81 KiB
Text
Generated
1650 lines
81 KiB
Text
Generated
{
|
|
"cells": [
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"# Kapitel 22: Praxisfallen und der Weg zum produktiven Einsatz\n",
|
|
"\n",
|
|
"Begleitnotebook zu *Optimierte Entscheidungsfindung mit Python*. Die Codezellen sind identisch mit den im Buch abgedruckten Programmen.\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"# Einmalig ausfuehren: installiert alle im Buch verwendeten Pakete.\n",
|
|
"# Lokal in einer virtuellen Umgebung genauso gueltig wie in Google Colab.\n",
|
|
"%pip install --quiet ortools highspy cvxpy scipy numpy pandas polars \\\n",
|
|
" scikit-learn matplotlib plotly pyomo linopy pymoo pydantic openpyxl"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Falle 1 — Infeasibility\n",
|
|
"\n",
|
|
"`Infeasibility_Diagnose.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Infeasibility_Diagnose.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Praxisfallen: Aus INFEASIBLE einen Notfallplan machen.\n",
|
|
"\n",
|
|
"Demonstriert die hierarchische Relaxation an einem Dienstplan, der in der\n",
|
|
"harten Fassung unloesbar ist, und zeigt, wie der Solver die Ursache benennt.\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from ortools.sat.python import cp_model\n",
|
|
"\n",
|
|
"SLOTS = [\"Mo frueh\", \"Mo spaet\", \"Di frueh\", \"Di spaet\"]\n",
|
|
"FAECHER = [\"Mathematik\", \"Physik\", \"Mathematik\", \"Chemie\"] # Chemie kann niemand!\n",
|
|
"PERSONAL = [\"Alice\", \"Bob\", \"Carla\"]\n",
|
|
"QUALIFIKATION = {\n",
|
|
" \"Alice\": {\"Mathematik\", \"Physik\"},\n",
|
|
" \"Bob\": {\"Mathematik\"},\n",
|
|
" \"Carla\": {\"Physik\", \"Mathematik\"},\n",
|
|
"}\n",
|
|
"MAX_PRO_PERSON = 2\n",
|
|
"STRAFE_UNBESETZT = 10_000 # sehr teuer, aber nicht unmoeglich\n",
|
|
"STRAFE_UEBERLAST = 500 # teuer, aber billiger als Ausfall\n",
|
|
"\n",
|
|
"\n",
|
|
"def plane(harte_fassung: bool):\n",
|
|
" \"\"\"harte_fassung=True: klassisch (kann INFEASIBLE werden).\n",
|
|
" harte_fassung=False: mit Schlupfvariablen (immer loesbar).\"\"\"\n",
|
|
" modell = cp_model.CpModel()\n",
|
|
" x = {(p, s): modell.NewBoolVar(f\"x_{p}_{s}\")\n",
|
|
" for p in PERSONAL for s in range(len(SLOTS))}\n",
|
|
"\n",
|
|
" strafen = []\n",
|
|
" unbesetzt = {}\n",
|
|
" ueberlast = {}\n",
|
|
"\n",
|
|
" for s in range(len(SLOTS)):\n",
|
|
" if harte_fassung:\n",
|
|
" modell.AddExactlyOne(x[p, s] for p in PERSONAL)\n",
|
|
" else:\n",
|
|
" # Schlupfvariable: der Slot DARF unbesetzt bleiben - gegen hohe Strafe\n",
|
|
" unbesetzt[s] = modell.NewBoolVar(f\"unbesetzt_{s}\")\n",
|
|
" modell.AddExactlyOne([x[p, s] for p in PERSONAL] + [unbesetzt[s]])\n",
|
|
" strafen.append(unbesetzt[s] * STRAFE_UNBESETZT)\n",
|
|
"\n",
|
|
" # Qualifikation bleibt IMMER hart - fachfremder Unterricht ist keine Option\n",
|
|
" for s, fach in enumerate(FAECHER):\n",
|
|
" for p in PERSONAL:\n",
|
|
" if fach not in QUALIFIKATION[p]:\n",
|
|
" modell.Add(x[p, s] == 0)\n",
|
|
"\n",
|
|
" for p in PERSONAL:\n",
|
|
" last = sum(x[p, s] for s in range(len(SLOTS)))\n",
|
|
" if harte_fassung:\n",
|
|
" modell.Add(last <= MAX_PRO_PERSON)\n",
|
|
" else:\n",
|
|
" # Ueberlast erlaubt - aber teuer\n",
|
|
" ueberlast[p] = modell.NewIntVar(0, len(SLOTS), f\"ueberlast_{p}\")\n",
|
|
" modell.Add(last <= MAX_PRO_PERSON + ueberlast[p])\n",
|
|
" strafen.append(ueberlast[p] * STRAFE_UEBERLAST)\n",
|
|
"\n",
|
|
" if strafen:\n",
|
|
" modell.Minimize(sum(strafen))\n",
|
|
"\n",
|
|
" loeser = cp_model.CpSolver()\n",
|
|
" loeser.parameters.max_time_in_seconds = 5.0\n",
|
|
" status = loeser.Solve(modell)\n",
|
|
" return loeser, status, x, unbesetzt, ueberlast\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\" VARIANTE A: KLASSISCH MIT LAUTER HARTEN BEDINGUNGEN\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" loeser, status, *_ = plane(harte_fassung=True)\n",
|
|
" print(f\"Solver-Status: {loeser.StatusName(status)}\")\n",
|
|
" if status == cp_model.INFEASIBLE:\n",
|
|
" print(\"Das System kann keinen Plan liefern. Der Anwender erfaehrt NICHT,\")\n",
|
|
" print(\"welche Regel das Problem verursacht - nur, dass es nicht geht.\\n\")\n",
|
|
"\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\" VARIANTE B: MIT HIERARCHISCHER RELAXATION\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" loeser, status, x, unbesetzt, ueberlast = plane(harte_fassung=False)\n",
|
|
" print(f\"Solver-Status: {loeser.StatusName(status)} | \"\n",
|
|
" f\"Strafkosten: {loeser.ObjectiveValue():.0f}\\n\")\n",
|
|
"\n",
|
|
" print(f\"{'Slot':<12} {'Fach':<12} {'Zuweisung':<18} {'Bemerkung'}\")\n",
|
|
" print(\"-\" * 78)\n",
|
|
" for s, slot in enumerate(SLOTS):\n",
|
|
" if loeser.Value(unbesetzt[s]):\n",
|
|
" print(f\"{slot:<12} {FAECHER[s]:<12} {'-- UNBESETZT --':<18} \"\n",
|
|
" f\"Kein qualifiziertes Personal verfuegbar\")\n",
|
|
" else:\n",
|
|
" person = next(p for p in PERSONAL if loeser.Value(x[p, s]))\n",
|
|
" print(f\"{slot:<12} {FAECHER[s]:<12} {person:<18}\")\n",
|
|
"\n",
|
|
" print(\"\\n--- Diagnose ---\")\n",
|
|
" for s, slot in enumerate(SLOTS):\n",
|
|
" if loeser.Value(unbesetzt[s]):\n",
|
|
" fach = FAECHER[s]\n",
|
|
" qualifiziert = [p for p in PERSONAL if fach in QUALIFIKATION[p]]\n",
|
|
" print(f\" '{slot}' ({fach}): {len(qualifiziert)} qualifizierte Personen \"\n",
|
|
" f\"{qualifiziert if qualifiziert else '-> URSACHE: niemand kann dieses Fach'}\")\n",
|
|
" for p in PERSONAL:\n",
|
|
" if loeser.Value(ueberlast[p]):\n",
|
|
" print(f\" {p} arbeitet {loeser.Value(ueberlast[p])} Stunden ueber der Grenze.\")\n",
|
|
"\n",
|
|
" print(\"\\nDer Anwender bekommt jetzt einen Plan PLUS eine konkrete Ursache -\")\n",
|
|
" print(\"und kann handeln: Vertretung von aussen holen, Stunde verlegen,\")\n",
|
|
" print(\"oder die Klasse zusammenlegen.\")\n",
|
|
" print(\"=\" * 78)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Falle 2 — Der Black-Box-Effekt\n",
|
|
"\n",
|
|
"`Erklaerbarkeit.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Erklaerbarkeit.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Praxisfallen: Ein Constraint-Trace, der die drei Fragen der Faustregel fuer\n",
|
|
"Erklaerbarkeit (Abschnitt 'Die fuenf typischen Praxisfallen') fuer eine\n",
|
|
"einzelne Zuweisung beantwortet:\n",
|
|
"Warum diese Loesung? Warum nicht die Alternative? Was wuerde sie verbessern?\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from ortools.sat.python import cp_model\n",
|
|
"\n",
|
|
"PERSONAL = [\"Anna\", \"Ben\", \"Clara\"]\n",
|
|
"BEREITS_STUNDEN = {\"Anna\": 1, \"Ben\": 2, \"Clara\": 0} # diese Woche schon geleistet\n",
|
|
"MAX_STUNDEN = 2 # gesetzliche Obergrenze\n",
|
|
"STRAFE_VORBELASTUNG = 50 # pro bereits geleisteter Stunde\n",
|
|
"STRAFKOSTEN_WUNSCH = {\"Anna\": 0, \"Ben\": 30, \"Clara\": 100}\n",
|
|
"\n",
|
|
"\n",
|
|
"def loese(max_stunden: int):\n",
|
|
" modell = cp_model.CpModel()\n",
|
|
" x = {p: modell.NewBoolVar(f\"x_{p}\") for p in PERSONAL}\n",
|
|
" modell.AddExactlyOne(x.values())\n",
|
|
"\n",
|
|
" kosten = []\n",
|
|
" for p in PERSONAL:\n",
|
|
" if BEREITS_STUNDEN[p] + 1 > max_stunden:\n",
|
|
" modell.Add(x[p] == 0) # hartes Ausschlusskriterium\n",
|
|
" strafe = BEREITS_STUNDEN[p] * STRAFE_VORBELASTUNG + STRAFKOSTEN_WUNSCH[p]\n",
|
|
" kosten.append(strafe * x[p])\n",
|
|
"\n",
|
|
" modell.Minimize(sum(kosten))\n",
|
|
" loeser = cp_model.CpSolver()\n",
|
|
" status = loeser.Solve(modell)\n",
|
|
" if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):\n",
|
|
" return None\n",
|
|
" gewinner = next(p for p in PERSONAL if loeser.Value(x[p]))\n",
|
|
" return gewinner, loeser.ObjectiveValue()\n",
|
|
"\n",
|
|
"\n",
|
|
"def gesamtkosten(p: str) -> int:\n",
|
|
" return BEREITS_STUNDEN[p] * STRAFE_VORBELASTUNG + STRAFKOSTEN_WUNSCH[p]\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" gewinner, kosten_opt = loese(MAX_STUNDEN)\n",
|
|
"\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\" CONSTRAINT-TRACE: Warum bekommt\", gewinner, \"die Vertretungsstunde?\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
"\n",
|
|
" print(\"\\n1. WARUM DIESE LOESUNG?\")\n",
|
|
" v, w = BEREITS_STUNDEN[gewinner] * STRAFE_VORBELASTUNG, STRAFKOSTEN_WUNSCH[gewinner]\n",
|
|
" print(f\" {gewinner}: Vorbelastung {v} Strafpunkte + Wunsch {w} Strafpunkte \"\n",
|
|
" f\"= {v + w} Strafpunkte insgesamt (Minimum).\")\n",
|
|
"\n",
|
|
" print(\"\\n2. WARUM NICHT DIE NAHELIEGENDEN ALTERNATIVEN?\")\n",
|
|
" for p in PERSONAL:\n",
|
|
" if p == gewinner:\n",
|
|
" continue\n",
|
|
" if BEREITS_STUNDEN[p] + 1 > MAX_STUNDEN:\n",
|
|
" print(f\" {p}: AUSGESCHLOSSEN - hat bereits {BEREITS_STUNDEN[p]} Stunden, \"\n",
|
|
" f\"eine dritte würde die Obergrenze von {MAX_STUNDEN} Stunden verletzen \"\n",
|
|
" f\"(harte Bedingung, nicht verhandelbar).\")\n",
|
|
" else:\n",
|
|
" differenz = gesamtkosten(p) - kosten_opt\n",
|
|
" print(f\" {p}: zulässig, aber {differenz:.0f} Strafpunkte teurer als \"\n",
|
|
" f\"{gewinner} ({gesamtkosten(p):.0f} statt {kosten_opt:.0f}).\")\n",
|
|
"\n",
|
|
" print(\"\\n3. WAS WUERDE DIE LOESUNG VERBESSERN?\")\n",
|
|
" gewinner_gelockert, kosten_gelockert = loese(MAX_STUNDEN + 1)\n",
|
|
" if kosten_gelockert < kosten_opt:\n",
|
|
" print(f\" Obergrenze auf {MAX_STUNDEN + 1} Stunden gelockert: {gewinner_gelockert} \"\n",
|
|
" f\"uebernimmt jetzt zu {kosten_gelockert:.0f} Strafpunkten \"\n",
|
|
" f\"({kosten_opt - kosten_gelockert:.0f} weniger als heute).\")\n",
|
|
" print(\" -> Diese Bedingung hat tatsaechlich einen Preis.\")\n",
|
|
" else:\n",
|
|
" print(f\" Obergrenze auf {MAX_STUNDEN + 1} Stunden gelockert: Ergebnis bleibt bei \"\n",
|
|
" f\"{gewinner} mit {kosten_gelockert:.0f} Strafpunkten (unveraendert).\")\n",
|
|
" blockiert = \", \".join(p for p in PERSONAL\n",
|
|
" if BEREITS_STUNDEN[p] + 1 > MAX_STUNDEN and p != gewinner)\n",
|
|
" print(f\" -> Die Obergrenze blockiert zwar {blockiert}, kostet aber gerade NICHTS:\")\n",
|
|
" print(\" Selbst ohne sie waere die Loesung dieselbe.\")\n",
|
|
" print(\" Blockierend und kostenrelevant sind zwei verschiedene Dinge.\")\n",
|
|
" print(\"=\" * 78)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Constraint Attribution: welche Bedingung kostet wie viel?\n",
|
|
"\n",
|
|
"`Constraint_Attribution.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Constraint_Attribution.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Praxisfallen: Welche Bedingung kostet wie viel - und was sagt man\n",
|
|
"dem Management?\n",
|
|
"\n",
|
|
"Zwischen den beiden bekannten Faellen klafft eine Luecke:\n",
|
|
"\n",
|
|
" * Erklaerbarkeit.py beantwortet \"warum diese eine Zuweisung?\"\n",
|
|
" * Konfliktsuche.py (Anhang Fehlerdiagnose) beantwortet \"warum geht es gar\n",
|
|
" nicht?\"\n",
|
|
"\n",
|
|
"Der haeufigste Fall in der Praxis liegt dazwischen: Das Modell rechnet, der\n",
|
|
"Plan ist zulaessig - und trotzdem unbefriedigend. Niemand weiss, WORAN es\n",
|
|
"liegt. Genau das beantwortet Constraint Attribution.\n",
|
|
"\n",
|
|
"Gerechnet wird auf derselben Fabrik wie in Konfliktsuche.py, nur nach der\n",
|
|
"Reparatur: Dort war die Lackierkapazitaet von 150 Stunden der Kern des\n",
|
|
"Widerspruchs (2*40 + 3*30 = 170 > 150). Eine zweite Schicht hat sie auf 210\n",
|
|
"gebracht, und der Mindestumsatz wurde auf 10.000 EUR gesenkt. Jetzt ist das\n",
|
|
"Modell loesbar - und die Frage lautet nicht mehr \"warum nicht?\", sondern\n",
|
|
"\"was kostet uns was?\".\n",
|
|
"\n",
|
|
"Gezeigt werden vier Dinge:\n",
|
|
" 1. Welche Bedingungen binden - und dass \"bindend\" nicht \"teuer\" heisst.\n",
|
|
" 2. Dass der Schattenpreis eine MOMENTAUFNAHME ist: hochgerechnet auf eine\n",
|
|
" realistische Sonderschicht verspricht er hier das Doppelte des\n",
|
|
" tatsaechlichen Nutzens.\n",
|
|
" 3. Dass ein Wunsch aus dem Vertrieb nicht teuer, sondern unmoeglich sein\n",
|
|
" kann - und welche Bedingungen ihn gemeinsam blockieren.\n",
|
|
" 4. Wie aus alledem ein Bericht in Alltagssprache wird.\n",
|
|
"\n",
|
|
"Benoetigt: numpy, scipy\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from __future__ import annotations\n",
|
|
"\n",
|
|
"import textwrap\n",
|
|
"\n",
|
|
"import numpy as np\n",
|
|
"from scipy.optimize import linprog\n",
|
|
"\n",
|
|
"PRODUKTE = [\"Rahmen\", \"Gehaeuse\", \"Deckel\", \"Traeger\", \"Halter\"]\n",
|
|
"DECKUNGSBEITRAG = np.array([35.0, 28.0, 9.0, 22.0, 6.0]) # EUR je Stueck\n",
|
|
"\n",
|
|
"# Dieselben Bedingungen wie in Konfliktsuche.py, mit den beiden Reparaturen.\n",
|
|
"# Der sprechende Name ist keine Kosmetik: Ohne ihn ist kein Bericht moeglich,\n",
|
|
"# der jemandem ausserhalb der Modellierung etwas sagt.\n",
|
|
"BEDINGUNGEN: list[tuple[str, list[float], str, float]] = [\n",
|
|
" (\"Kapazitaet Montage\", [3, 2, 1, 4, 2], \"<=\", 400),\n",
|
|
" (\"Kapazitaet Lackieren\", [2, 3, 0, 1, 0], \"<=\", 210),\n",
|
|
" (\"Kapazitaet Pruefung\", [1, 1, 1, 1, 1], \"<=\", 300),\n",
|
|
" (\"Liefervertrag Rahmen\", [1, 0, 0, 0, 0], \">=\", 40),\n",
|
|
" (\"Liefervertrag Gehaeuse\", [0, 1, 0, 0, 0], \">=\", 30),\n",
|
|
" (\"Liefervertrag Deckel\", [0, 0, 1, 0, 0], \">=\", 20),\n",
|
|
" (\"Liefervertrag Traeger\", [0, 0, 0, 1, 0], \">=\", 25),\n",
|
|
" (\"Marktgrenze Rahmen\", [1, 0, 0, 0, 0], \"<=\", 120),\n",
|
|
" (\"Marktgrenze Gehaeuse\", [0, 1, 0, 0, 0], \"<=\", 90),\n",
|
|
" (\"Marktgrenze Deckel\", [0, 0, 1, 0, 0], \"<=\", 150),\n",
|
|
" (\"Marktgrenze Halter\", [0, 0, 0, 0, 1], \"<=\", 200),\n",
|
|
" (\"Mindestumsatz\", [90, 70, 30, 60, 20], \">=\", 10000),\n",
|
|
" (\"Sortimentsbreite\", [0, 0, 1, 1, 1], \">=\", 100),\n",
|
|
" (\"Lackierbudget Schicht 2\", [0, 1, 0, 1, 0], \"<=\", 55),\n",
|
|
"]\n",
|
|
"\n",
|
|
"# Was liesse sich im Ernstfall wirklich bewegen - und um wie viel? Nicht \"eine\n",
|
|
"# Einheit\", sondern der Schritt, den ein Planer tatsaechlich gehen kann: eine\n",
|
|
"# Sonderschicht, ein nachverhandelter Vertrag. Genau an dieser Groesse\n",
|
|
"# scheitert das blosse Hochrechnen des Schattenpreises.\n",
|
|
"STELLSCHRAUBEN: dict[str, tuple[float, str]] = {\n",
|
|
" \"Kapazitaet Montage\": (+50, \"eine Sonderschicht Montage\"),\n",
|
|
" \"Kapazitaet Lackieren\": (+30, \"eine Sonderschicht Lackieren\"),\n",
|
|
" \"Kapazitaet Pruefung\": (+50, \"eine Sonderschicht Pruefung\"),\n",
|
|
" \"Lackierbudget Schicht 2\": (+10, \"10 Stunden mehr Budget in Schicht 2\"),\n",
|
|
" \"Liefervertrag Traeger\": (-10, \"10 Stueck weniger Abnahmepflicht Traeger\"),\n",
|
|
" \"Liefervertrag Gehaeuse\": (-10, \"10 Stueck weniger Abnahmepflicht Gehaeuse\"),\n",
|
|
"}\n",
|
|
"\n",
|
|
"WUNSCH_MENGE = 55.0 # der Vertrieb haette gern mehr Rahmen\n",
|
|
"WUNSCH_PRODUKT = \"Rahmen\"\n",
|
|
"\n",
|
|
"\n",
|
|
"def loese(bedingungen: list[tuple[str, list[float], str, float]]):\n",
|
|
" \"\"\"Maximiert den Deckungsbeitrag. linprog minimiert, daher das Vorzeichen.\"\"\"\n",
|
|
" matrix, rechte_seite = [], []\n",
|
|
" for _, koeffizienten, richtung, grenze in bedingungen:\n",
|
|
" zeile = np.array(koeffizienten, dtype=float)\n",
|
|
" matrix.append(zeile if richtung == \"<=\" else -zeile)\n",
|
|
" rechte_seite.append(grenze if richtung == \"<=\" else -grenze)\n",
|
|
" return linprog(-DECKUNGSBEITRAG,\n",
|
|
" A_ub=np.array(matrix), b_ub=np.array(rechte_seite),\n",
|
|
" bounds=[(0, None)] * len(PRODUKTE), method=\"highs\")\n",
|
|
"\n",
|
|
"\n",
|
|
"def geaendert(name: str, delta: float,\n",
|
|
" basis=None) -> list[tuple[str, list[float], str, float]]:\n",
|
|
" \"\"\"Dieselben Bedingungen, eine davon um delta verschoben.\"\"\"\n",
|
|
" return [(n, k, r, g + delta if n == name else g)\n",
|
|
" for n, k, r, g in (basis or BEDINGUNGEN)]\n",
|
|
"\n",
|
|
"\n",
|
|
"def auswertung(ergebnis, bedingungen=None):\n",
|
|
" \"\"\"Schlupf und Schattenpreis je Bedingung.\"\"\"\n",
|
|
" x = ergebnis.x\n",
|
|
" dual = ergebnis.ineqlin.marginals\n",
|
|
" zeilen = []\n",
|
|
" for i, (name, koeff, richtung, grenze) in enumerate(bedingungen or BEDINGUNGEN):\n",
|
|
" links = float(np.array(koeff, dtype=float) @ x)\n",
|
|
" schlupf = (grenze - links) if richtung == \"<=\" else (links - grenze)\n",
|
|
" zeilen.append((name, links, richtung, grenze, schlupf, abs(dual[i])))\n",
|
|
" return zeilen\n",
|
|
"\n",
|
|
"\n",
|
|
"def blockierer(bedingungen, zusatz) -> list[str]:\n",
|
|
" \"\"\"Welche Bedingungen verhindern 'zusatz' - einzeln weggelassen.\n",
|
|
"\n",
|
|
" Dasselbe Vorgehen wie der Deletion Filter in Konfliktsuche.py, nur mit\n",
|
|
" umgekehrtem Vorzeichen: Dort wird gesucht, was den Widerspruch AUSMACHT,\n",
|
|
" hier, was einen Wunsch VERHINDERT. Wird das Modell ohne eine Bedingung\n",
|
|
" loesbar, gehoert sie zum Konflikt.\n",
|
|
" \"\"\"\n",
|
|
" gefunden = []\n",
|
|
" for i, (name, _, _, _) in enumerate(bedingungen):\n",
|
|
" rest = [c for j, c in enumerate(bedingungen) if j != i] + [zusatz]\n",
|
|
" if loese(rest).success:\n",
|
|
" gefunden.append(name)\n",
|
|
" return gefunden\n",
|
|
"\n",
|
|
"\n",
|
|
"def euro(betrag: float) -> str:\n",
|
|
" return f\"{betrag:,.2f} EUR\"\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" basis = loese(BEDINGUNGEN)\n",
|
|
" db_basis = -basis.fun\n",
|
|
" plan = dict(zip(PRODUKTE, basis.x))\n",
|
|
"\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\" CONSTRAINT ATTRIBUTION: Was kostet uns welche Bedingung?\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(f\"Status: {basis.message.split('(')[0].strip()}\")\n",
|
|
" print(f\"Deckungsbeitrag des Plans: {euro(db_basis)}\\n\")\n",
|
|
" print(\" \" + \" \".join(f\"{p}: {m:.1f}\" for p, m in plan.items()))\n",
|
|
"\n",
|
|
" # --- 1. Wer begrenzt den Plan? -----------------------------------------\n",
|
|
" print(\"\\n\" + \"=\" * 78)\n",
|
|
" print(\" (1) Welche Bedingungen binden - und was kostet die naechste Einheit?\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(f\"{'Bedingung':<24} {'genutzt':>9} {'Grenze':>7} {'Schlupf':>8} \"\n",
|
|
" f\"{'Schattenpreis':>11}\")\n",
|
|
" print(\"-\" * 78)\n",
|
|
" bindend = []\n",
|
|
" for name, links, richtung, grenze, schlupf, dual in auswertung(basis):\n",
|
|
" marke = \"\"\n",
|
|
" if abs(schlupf) < 1e-6:\n",
|
|
" bindend.append((name, dual))\n",
|
|
" marke = \" <-- bindend\"\n",
|
|
" print(f\"{name:<24} {links:>9.1f} {grenze:>7.0f} {schlupf:>8.1f} \"\n",
|
|
" f\"{dual:>11.2f}{marke}\")\n",
|
|
"\n",
|
|
" ohne_preis = [n for n, d in bindend if d < 1e-9]\n",
|
|
" print(f\"\\n {len(bindend)} Bedingungen binden. Aber: \"\n",
|
|
" f\"{', '.join(ohne_preis)} bindet\")\n",
|
|
" print(\" mit einem Schattenpreis von 0,00 EUR - sie beruehrt den Plan, ohne ihn\")\n",
|
|
" print(\" zu verteuern. Bindend und kostenrelevant sind zwei verschiedene Dinge.\")\n",
|
|
"\n",
|
|
" # --- 2. Der Schattenpreis als Momentaufnahme ---------------------------\n",
|
|
" print(\"\\n\" + \"=\" * 78)\n",
|
|
" print(\" (2) Was der Schattenpreis verschweigt\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\"Der Schattenpreis gilt fuer die NAECHSTE Einheit. Ein Planer kauft aber\")\n",
|
|
" print(\"keine Einheit, sondern eine ganze Schicht. Beides gegenuebergestellt:\\n\")\n",
|
|
" preise = {name: dual for name, *_, dual in auswertung(basis)}\n",
|
|
" print(f\"{'Stellschraube':<24} {'Preis':>9} {'Schritt':>7} \"\n",
|
|
" f\"{'hochgerechnet':>13} {'gemessen':>9} {'Abw.':>9}\")\n",
|
|
" print(\"-\" * 78)\n",
|
|
" wirkung = {}\n",
|
|
" for name, (delta, _) in STELLSCHRAUBEN.items():\n",
|
|
" ergebnis = loese(geaendert(name, delta))\n",
|
|
" echt = (-ergebnis.fun) - db_basis if ergebnis.success else 0.0\n",
|
|
" linear = preise[name] * abs(delta)\n",
|
|
" wirkung[name] = echt\n",
|
|
" print(f\"{name:<24} {preise[name]:>9.2f} {delta:>+7.0f} \"\n",
|
|
" f\"{linear:>13.2f} {echt:>9.2f} {echt - linear:>+9.2f}\")\n",
|
|
"\n",
|
|
" # Wo endet der Gueltigkeitsbereich? Schrittweise nachfahren.\n",
|
|
" abweichler = max(STELLSCHRAUBEN,\n",
|
|
" key=lambda n: preise[n] * abs(STELLSCHRAUBEN[n][0]) - wirkung[n])\n",
|
|
" delta_max = STELLSCHRAUBEN[abweichler][0]\n",
|
|
" grenze_gueltig = 0.0\n",
|
|
" schritt = delta_max / 60.0\n",
|
|
" vorher = db_basis\n",
|
|
" for i in range(1, 61):\n",
|
|
" d = schritt * i\n",
|
|
" e = loese(geaendert(abweichler, d))\n",
|
|
" db = -e.fun if e.success else vorher\n",
|
|
" if db - vorher > 1e-9:\n",
|
|
" grenze_gueltig = d\n",
|
|
" vorher = db\n",
|
|
" print(f\"\\n '{abweichler}': hochgerechnet \"\n",
|
|
" f\"{preise[abweichler] * abs(delta_max):.2f} EUR, tatsaechlich \"\n",
|
|
" f\"{wirkung[abweichler]:.2f} EUR.\")\n",
|
|
" print(f\" Der Preis von {preise[abweichler]:.2f} EUR gilt nur bis \"\n",
|
|
" f\"+{grenze_gueltig:.0f} Stunden. Jede weitere Stunde\")\n",
|
|
" print(\" dieser Sonderschicht bringt exakt nichts - der Engpass ist dann ein anderer.\")\n",
|
|
"\n",
|
|
" nach_preis = sorted(STELLSCHRAUBEN, key=lambda n: -preise[n])[0]\n",
|
|
" nach_wirkung = max(wirkung, key=wirkung.get)\n",
|
|
" print(f\"\\n Und die Ranglisten drehen sich: Nach Schattenpreis fuehrt\")\n",
|
|
" print(f\" '{nach_preis}' ({preise[nach_preis]:.2f} EUR je Einheit),\")\n",
|
|
" print(f\" nach tatsaechlicher Wirkung '{nach_wirkung}' \"\n",
|
|
" f\"({wirkung[nach_wirkung]:.2f} EUR).\")\n",
|
|
" print(\" Der Schattenpreis sagt, was eine Einheit wert ist - nicht, wie viele\")\n",
|
|
" print(\" Einheiten zu haben sind.\")\n",
|
|
"\n",
|
|
" # --- 3. Der Wunsch aus dem Vertrieb ------------------------------------\n",
|
|
" print(\"\\n\" + \"=\" * 78)\n",
|
|
" print(f\" (3) Was-waere-wenn: mindestens {WUNSCH_MENGE:.0f} \"\n",
|
|
" f\"{WUNSCH_PRODUKT} statt {plan[WUNSCH_PRODUKT]:.1f}\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
" wunsch = (f\"Wunsch Vertrieb {WUNSCH_PRODUKT}\",\n",
|
|
" [1 if p == WUNSCH_PRODUKT else 0 for p in PRODUKTE], \">=\", WUNSCH_MENGE)\n",
|
|
" mit_wunsch = loese(BEDINGUNGEN + [wunsch])\n",
|
|
" schuldige: list[str] = []\n",
|
|
" engpass = noetig = None\n",
|
|
" wunsch_gratis = False\n",
|
|
"\n",
|
|
" if mit_wunsch.success:\n",
|
|
" print(f\"Erfuellbar, kostet {euro(db_basis - (-mit_wunsch.fun))}.\")\n",
|
|
" else:\n",
|
|
" schuldige = blockierer(BEDINGUNGEN, wunsch)\n",
|
|
" print(\"Der Wunsch ist nicht teuer - er ist UNMOEGLICH.\\n\")\n",
|
|
" print(f\"Blockiert wird er von {len(schuldige)} Bedingungen gemeinsam. \"\n",
|
|
" f\"Faellt eine davon\")\n",
|
|
" print(\"weg, ist er erfuellbar; keine allein ist 'der' Grund:\")\n",
|
|
" for name in schuldige:\n",
|
|
" e = loese([c for c in BEDINGUNGEN if c[0] != name] + [wunsch])\n",
|
|
" print(f\" ohne '{name}': moeglich, Deckungsbeitrag {euro(-e.fun)}\")\n",
|
|
"\n",
|
|
" engpass = next((n for n in schuldige if n in STELLSCHRAUBEN), schuldige[0])\n",
|
|
" for schritt in range(1, 61):\n",
|
|
" if loese(geaendert(engpass, schritt) + [wunsch]).success:\n",
|
|
" noetig = schritt\n",
|
|
" break\n",
|
|
" if noetig is not None:\n",
|
|
" e = loese(geaendert(engpass, noetig) + [wunsch])\n",
|
|
" ohne = loese(geaendert(engpass, noetig))\n",
|
|
" wunsch_gratis = abs((-e.fun) - (-ohne.fun)) < 1e-6\n",
|
|
" print(f\"\\n Mit +{noetig} bei '{engpass}' wird der Wunsch erfuellbar:\")\n",
|
|
" print(f\" Deckungsbeitrag {euro(-e.fun)} \"\n",
|
|
" f\"({-e.fun - db_basis:+,.2f} gegenueber heute).\")\n",
|
|
" if wunsch_gratis:\n",
|
|
" print(f\" Und ohne den Wunsch ergaebe dieselbe Lockerung \"\n",
|
|
" f\"{euro(-ohne.fun)} bei\")\n",
|
|
" print(f\" {ohne.x[PRODUKTE.index(WUNSCH_PRODUKT)]:.0f} \"\n",
|
|
" f\"{WUNSCH_PRODUKT} - der Wunsch kostet also NICHTS.\")\n",
|
|
" print(\" Er war nie das Problem. Der Engpass war es.\")\n",
|
|
"\n",
|
|
" # --- 4. Der Bericht ----------------------------------------------------\n",
|
|
" # Aus denselben Zahlen, die oben in Tabellen stehen, wird hier Prosa. Kein\n",
|
|
" # Satz enthaelt eine Zahl, die nicht vorher gerechnet wurde - das ist der\n",
|
|
" # ganze Trick und zugleich die Bedingung dafuer, dass der Bericht nicht\n",
|
|
" # irgendwann leise falsch wird.\n",
|
|
" print(\"\\n\" + \"=\" * 78)\n",
|
|
" print(\" (4) Und so steht es in der Vorlage fuer die Abteilungsleiterrunde\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
"\n",
|
|
" teuerste = sorted(((n, d) for n, d in bindend if d > 0), key=lambda t: -t[1])\n",
|
|
" gross = STELLSCHRAUBEN[nach_wirkung][1]\n",
|
|
" klein = STELLSCHRAUBEN[abweichler][1]\n",
|
|
"\n",
|
|
" absaetze = [\n",
|
|
" f\"Der Plan bringt {euro(db_basis)} Deckungsbeitrag. Begrenzt wird er von \"\n",
|
|
" f\"{len(bindend)} Bedingungen, von denen {len(teuerste)} tatsaechlich Geld \"\n",
|
|
" f\"kosten; \\\"{ohne_preis[0]}\\\" beruehrt den Plan, ohne ihn zu verteuern.\",\n",
|
|
"\n",
|
|
" f\"Teuerste Bindung ist \\\"{teuerste[0][0]}\\\" mit \"\n",
|
|
" f\"{teuerste[0][1]:.2f} EUR je Einheit, gefolgt von \\\"{teuerste[1][0]}\\\" \"\n",
|
|
" f\"mit {teuerste[1][1]:.2f} EUR.\",\n",
|
|
"\n",
|
|
" f\"Die groesste einzelne Verbesserung bringt {gross}: \"\n",
|
|
" f\"{euro(wirkung[nach_wirkung])} mehr Deckungsbeitrag. {klein[0].upper()}{klein[1:]} \"\n",
|
|
" f\"dagegen bringt nur {euro(wirkung[abweichler])} statt der rechnerischen \"\n",
|
|
" f\"{euro(preise[abweichler] * abs(delta_max))} - ab +{grenze_gueltig:.0f} \"\n",
|
|
" f\"Stunden begrenzt uns etwas anderes. Wer die ganze Schicht bezahlt, \"\n",
|
|
" f\"bezahlt die zweite Haelfte umsonst.\",\n",
|
|
" ]\n",
|
|
"\n",
|
|
" if schuldige and noetig is not None:\n",
|
|
" namen = \", \".join(f'\"{s}\"' for s in schuldige)\n",
|
|
" satz = (f\"Zum Wunsch des Vertriebs ({WUNSCH_MENGE:.0f} {WUNSCH_PRODUKT} statt \"\n",
|
|
" f\"{plan[WUNSCH_PRODUKT]:.1f}): Er ist mit der heutigen Ausstattung \"\n",
|
|
" f\"nicht erfuellbar - nicht aus Kostengruenden, sondern weil {namen} \"\n",
|
|
" f\"ihn gemeinsam blockieren. Mit +{noetig} Stunden bei \"\n",
|
|
" f'\"{engpass}\" wird er moeglich')\n",
|
|
" satz += (\", und er kostet dann nichts: Der Plan waehlt diese Menge von \"\n",
|
|
" \"sich aus.\" if wunsch_gratis else \".\")\n",
|
|
" absaetze.append(satz)\n",
|
|
"\n",
|
|
" for absatz in absaetze:\n",
|
|
" print()\n",
|
|
" print(textwrap.fill(absatz, width=76))\n",
|
|
" print(\"\\n\" + \"=\" * 78)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Ein Enum für fünf Bibliotheken\n",
|
|
"\n",
|
|
"`or_kern.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# or_kern.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Praxisfallen: Der gemeinsame Unterbau fuer produktionsreife OR-Modelle.\n",
|
|
"\n",
|
|
"Alle Beispiele dieses Buchs loesen dieselben vier Aufgaben immer wieder:\n",
|
|
"Daten einlesen und pruefen, ein Modell bauen, den Solverstatus auswerten, die\n",
|
|
"Loesung gegen die Wirklichkeit kontrollieren. Dieses Modul zieht diese vier\n",
|
|
"Aufgaben aus den Einzelprogrammen heraus.\n",
|
|
"\n",
|
|
" Rohdaten (Excel/CSV)\n",
|
|
" -> Domaenenmodell (Pydantic, geprueft)\n",
|
|
" -> Modellbauer (solverabhaengig)\n",
|
|
" -> Loesung (DTO, solverunabhaengig)\n",
|
|
" -> Abnahmepruefung\n",
|
|
"\n",
|
|
"Der Sinn dieser Trennung: Die Domaenenschicht weiss nichts von Solvern, die\n",
|
|
"Loesungsschicht weiss nichts vom Modellaufbau. Wer den Solver wechselt, tauscht\n",
|
|
"genau EINEN Baustein aus - der Rest bleibt.\n",
|
|
"\n",
|
|
"WICHTIG ZUM IMPORT: Dieses Modul importiert von sich aus KEINE Solver-\n",
|
|
"bibliothek. Der Grund steht im Kapitel Oekosystem: ortools und highspy vertragen sich\n",
|
|
"nicht im selben Prozess. Die Statusuebersetzer laden ihre Bibliothek erst,\n",
|
|
"wenn sie aufgerufen werden - so bleibt or_kern.py mit jedem Solver benutzbar.\n",
|
|
"\n",
|
|
"Benoetigt: pydantic (v2), numpy; optional pandas und openpyxl fuer Excel.\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from __future__ import annotations\n",
|
|
"\n",
|
|
"from enum import Enum\n",
|
|
"from typing import Annotated, Any, Sequence\n",
|
|
"\n",
|
|
"import numpy as np\n",
|
|
"from pydantic import BaseModel, Field, model_validator\n",
|
|
"\n",
|
|
"# Wiederverwendbare Feldtypen. Sie tragen die Pruefung im Typ, nicht im Code -\n",
|
|
"# damit gilt sie ueberall, wo der Typ verwendet wird.\n",
|
|
"PositiveZahl = Annotated[float, Field(gt=0)]\n",
|
|
"NichtNegativ = Annotated[float, Field(ge=0)]\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 1. Solverstatus: eine Sprache fuer fuenf Bibliotheken -------------------\n",
|
|
"\n",
|
|
"class SolverStatus(str, Enum):\n",
|
|
" \"\"\"Was ein Solverlauf ergeben hat - unabhaengig davon, wer gerechnet hat.\n",
|
|
"\n",
|
|
" Die fuenf im Buch verwendeten Bibliotheken benennen dasselbe Ergebnis\n",
|
|
" unterschiedlich: 'Optimal', 'OPTIMAL', 2, 'optimal'. Wer darauf direkt\n",
|
|
" prueft, bindet seinen Auswertungscode an eine Bibliothek.\n",
|
|
" \"\"\"\n",
|
|
" OPTIMAL = \"optimal\" # beweisbar bestmoeglich\n",
|
|
" ZULAESSIG = \"zulaessig\" # brauchbare Loesung, Beweis fehlt\n",
|
|
" UNZULAESSIG = \"unzulaessig\" # es gibt keine Loesung (Aussage ueber das Modell)\n",
|
|
" UNBESCHRAENKT = \"unbeschraenkt\" # Ziel waechst ins Unendliche\n",
|
|
" ZEITLIMIT = \"zeitlimit\" # abgebrochen, nichts Brauchbares gefunden\n",
|
|
" FEHLERHAFT = \"fehlerhaft\" # das Modell ist kein gueltiges Modell\n",
|
|
" UNBEKANNT = \"unbekannt\" # alles Uebrige\n",
|
|
"\n",
|
|
" @property\n",
|
|
" def brauchbar(self) -> bool:\n",
|
|
" \"\"\"Habe ich etwas in der Hand, das ich ausfuehren kann?\"\"\"\n",
|
|
" return self in (SolverStatus.OPTIMAL, SolverStatus.ZULAESSIG)\n",
|
|
"\n",
|
|
" @property\n",
|
|
" def modellfehler(self) -> bool:\n",
|
|
" \"\"\"Liegt die Ursache im Modell (und nicht in der Rechenzeit)?\"\"\"\n",
|
|
" return self in (SolverStatus.UNZULAESSIG, SolverStatus.UNBESCHRAENKT,\n",
|
|
" SolverStatus.FEHLERHAFT)\n",
|
|
"\n",
|
|
"\n",
|
|
"def status_von_pywraplp(rohstatus: int) -> SolverStatus:\n",
|
|
" \"\"\"OR-Tools linear_solver (GLOP, SCIP, CBC).\"\"\"\n",
|
|
" from ortools.linear_solver import pywraplp\n",
|
|
" zuordnung = {\n",
|
|
" pywraplp.Solver.OPTIMAL: SolverStatus.OPTIMAL,\n",
|
|
" pywraplp.Solver.FEASIBLE: SolverStatus.ZULAESSIG,\n",
|
|
" pywraplp.Solver.INFEASIBLE: SolverStatus.UNZULAESSIG,\n",
|
|
" pywraplp.Solver.UNBOUNDED: SolverStatus.UNBESCHRAENKT,\n",
|
|
" pywraplp.Solver.ABNORMAL: SolverStatus.FEHLERHAFT,\n",
|
|
" pywraplp.Solver.NOT_SOLVED: SolverStatus.ZEITLIMIT,\n",
|
|
" }\n",
|
|
" return zuordnung.get(rohstatus, SolverStatus.UNBEKANNT)\n",
|
|
"\n",
|
|
"\n",
|
|
"def status_von_cpsat(rohstatus: int) -> SolverStatus:\n",
|
|
" \"\"\"OR-Tools CP-SAT (siehe Kapitel CP-SAT, alle fuenf Faelle).\"\"\"\n",
|
|
" from ortools.sat.python import cp_model\n",
|
|
" zuordnung = {\n",
|
|
" cp_model.OPTIMAL: SolverStatus.OPTIMAL,\n",
|
|
" cp_model.FEASIBLE: SolverStatus.ZULAESSIG,\n",
|
|
" cp_model.INFEASIBLE: SolverStatus.UNZULAESSIG,\n",
|
|
" cp_model.MODEL_INVALID: SolverStatus.FEHLERHAFT,\n",
|
|
" cp_model.UNKNOWN: SolverStatus.ZEITLIMIT,\n",
|
|
" }\n",
|
|
" return zuordnung.get(rohstatus, SolverStatus.UNBEKANNT)\n",
|
|
"\n",
|
|
"\n",
|
|
"def status_von_highs(statustext: str) -> SolverStatus:\n",
|
|
" \"\"\"HiGHS ueber highspy - hier kommt der Status als Text.\"\"\"\n",
|
|
" zuordnung = {\n",
|
|
" \"Optimal\": SolverStatus.OPTIMAL,\n",
|
|
" \"Time limit reached\": SolverStatus.ZULAESSIG, # meist mit Loesung\n",
|
|
" \"Solution limit reached\": SolverStatus.ZULAESSIG,\n",
|
|
" \"Infeasible\": SolverStatus.UNZULAESSIG,\n",
|
|
" \"Unbounded\": SolverStatus.UNBESCHRAENKT,\n",
|
|
" \"Primal infeasible or unbounded\": SolverStatus.UNZULAESSIG,\n",
|
|
" \"Model error\": SolverStatus.FEHLERHAFT,\n",
|
|
" }\n",
|
|
" return zuordnung.get(statustext, SolverStatus.UNBEKANNT)\n",
|
|
"\n",
|
|
"\n",
|
|
"def status_von_scipy(ergebnis: Any) -> SolverStatus:\n",
|
|
" \"\"\"scipy.optimize.linprog und milp liefern ein Ergebnisobjekt mit .status.\"\"\"\n",
|
|
" zuordnung = {\n",
|
|
" 0: SolverStatus.OPTIMAL,\n",
|
|
" 1: SolverStatus.ZEITLIMIT, # Iterations-/Zeitgrenze\n",
|
|
" 2: SolverStatus.UNZULAESSIG,\n",
|
|
" 3: SolverStatus.UNBESCHRAENKT,\n",
|
|
" 4: SolverStatus.FEHLERHAFT, # numerische Schwierigkeiten\n",
|
|
" }\n",
|
|
" return zuordnung.get(int(ergebnis.status), SolverStatus.UNBEKANNT)\n",
|
|
"\n",
|
|
"\n",
|
|
"def status_von_cvxpy(statustext: str) -> SolverStatus:\n",
|
|
" \"\"\"CVXPY - Statuszeichenketten wie 'optimal' oder 'infeasible'.\"\"\"\n",
|
|
" zuordnung = {\n",
|
|
" \"optimal\": SolverStatus.OPTIMAL,\n",
|
|
" \"optimal_inaccurate\": SolverStatus.ZULAESSIG,\n",
|
|
" \"infeasible\": SolverStatus.UNZULAESSIG,\n",
|
|
" \"infeasible_inaccurate\": SolverStatus.UNZULAESSIG,\n",
|
|
" \"unbounded\": SolverStatus.UNBESCHRAENKT,\n",
|
|
" \"unbounded_inaccurate\": SolverStatus.UNBESCHRAENKT,\n",
|
|
" }\n",
|
|
" return zuordnung.get(statustext, SolverStatus.UNBEKANNT)\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 2. Die Loesung als solverunabhaengiges Datenobjekt ----------------------\n",
|
|
"\n",
|
|
"class Loesung(BaseModel):\n",
|
|
" \"\"\"Das Ergebnis eines Solverlaufs, so wie es weiterverarbeitet wird.\n",
|
|
"\n",
|
|
" Bewusst OHNE Referenz auf Solver, Modell oder Variablen: Ein Bericht, ein\n",
|
|
" Test oder eine Weiterverarbeitung soll nicht wissen muessen, womit\n",
|
|
" gerechnet wurde.\n",
|
|
" \"\"\"\n",
|
|
" status: SolverStatus\n",
|
|
" werte: dict[str, float] = Field(default_factory=dict)\n",
|
|
" zielwert: float | None = None\n",
|
|
" schranke: float | None = None\n",
|
|
" laufzeit: float | None = None\n",
|
|
" # Schattenpreise je Nebenbedingung, sofern der Solver sie liefert\n",
|
|
" # (nur bei kontinuierlichen Problemen, siehe Kapitel LP). Auch das ist\n",
|
|
" # ein solverunabhaengiger Begriff und gehoert deshalb hierher.\n",
|
|
" schattenpreise: dict[str, float] = Field(default_factory=dict)\n",
|
|
"\n",
|
|
" @property\n",
|
|
" def gap(self) -> float | None:\n",
|
|
" \"\"\"Relativer Abstand zwischen gefundener Loesung und bewiesener Schranke.\n",
|
|
"\n",
|
|
" Das ist die Zahl fuer den Bericht: keine Schaetzung, sondern eine\n",
|
|
" Zusage - schlechter als das kann die Loesung nicht sein.\n",
|
|
" \"\"\"\n",
|
|
" if self.zielwert is None or self.schranke is None:\n",
|
|
" return None\n",
|
|
" nenner = max(abs(self.zielwert), 1e-9)\n",
|
|
" return abs(self.zielwert - self.schranke) / nenner\n",
|
|
"\n",
|
|
" def als_bericht(self) -> str:\n",
|
|
" \"\"\"Eine Zeile fuers Betriebsprotokoll - siehe Betriebsueberwachung.py.\"\"\"\n",
|
|
" teile = [f\"Status: {self.status.value}\"]\n",
|
|
" if self.zielwert is not None:\n",
|
|
" teile.append(f\"Zielwert: {self.zielwert:,.2f}\")\n",
|
|
" if self.gap is not None:\n",
|
|
" teile.append(f\"Gap: {self.gap:.2%}\")\n",
|
|
" if self.laufzeit is not None:\n",
|
|
" teile.append(f\"Zeit: {self.laufzeit:.2f}s\")\n",
|
|
" return \" | \".join(teile)\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 3. Domaenenmodell: die Fakten, bevor ein Solver sie sieht ---------------\n",
|
|
"\n",
|
|
"class Produkt(BaseModel):\n",
|
|
" \"\"\"Ein Produkt mit Deckungsbeitrag und Ressourcenverbrauch.\"\"\"\n",
|
|
" name: str = Field(min_length=1)\n",
|
|
" deckungsbeitrag: float\n",
|
|
" verbrauch: dict[str, NichtNegativ]\n",
|
|
"\n",
|
|
" model_config = {\"frozen\": True} # Stammdaten aendern sich nicht im Lauf\n",
|
|
"\n",
|
|
"\n",
|
|
"class Produktionsproblem(BaseModel):\n",
|
|
" \"\"\"Produktionsprogrammplanung - das Modell aus Kapitel Einfuehrung.\n",
|
|
"\n",
|
|
" Die Pruefungen stehen hier und nicht im Solvercode. Damit gelten sie\n",
|
|
" unabhaengig davon, mit welcher Bibliothek spaeter gerechnet wird - und sie\n",
|
|
" schlagen beim EINLESEN zu, wo der Fehler noch zuzuordnen ist.\n",
|
|
" \"\"\"\n",
|
|
" produkte: list[Produkt] = Field(min_length=1)\n",
|
|
" kapazitaeten: dict[str, PositiveZahl]\n",
|
|
"\n",
|
|
" @model_validator(mode=\"after\")\n",
|
|
" def pruefe_ressourcen(self) -> \"Produktionsproblem\":\n",
|
|
" namen = [p.name for p in self.produkte]\n",
|
|
" if len(set(namen)) != len(namen):\n",
|
|
" doppelt = sorted({n for n in namen if namen.count(n) > 1})\n",
|
|
" raise ValueError(f\"Produktnamen kommen mehrfach vor: {doppelt}\")\n",
|
|
"\n",
|
|
" for produkt in self.produkte:\n",
|
|
" unbekannt = set(produkt.verbrauch) - set(self.kapazitaeten)\n",
|
|
" if unbekannt:\n",
|
|
" raise ValueError(\n",
|
|
" f\"Produkt '{produkt.name}' verbraucht Ressourcen ohne \"\n",
|
|
" f\"Kapazitaetsangabe: {sorted(unbekannt)}\")\n",
|
|
" return self\n",
|
|
"\n",
|
|
" @property\n",
|
|
" def ressourcen(self) -> list[str]:\n",
|
|
" \"\"\"Feste Reihenfolge - siehe die Spaltenfalle in Kapitel Finanzdaten.\"\"\"\n",
|
|
" return sorted(self.kapazitaeten)\n",
|
|
"\n",
|
|
" def verbrauchsmatrix(self) -> np.ndarray:\n",
|
|
" \"\"\"Zeilen = Ressourcen, Spalten = Produkte (Matrixform, Kapitel Fundament).\"\"\"\n",
|
|
" return np.array([[p.verbrauch.get(r, 0.0) for p in self.produkte]\n",
|
|
" for r in self.ressourcen])\n",
|
|
"\n",
|
|
" def kapazitaetsvektor(self) -> np.ndarray:\n",
|
|
" return np.array([self.kapazitaeten[r] for r in self.ressourcen])\n",
|
|
"\n",
|
|
" def deckungsbeitragsvektor(self) -> np.ndarray:\n",
|
|
" return np.array([p.deckungsbeitrag for p in self.produkte])\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 4. Abnahmepruefung: Loesung gegen Anforderung, ohne Solver --------------\n",
|
|
"\n",
|
|
"def pruefe_loesung(problem: Produktionsproblem, loesung: Loesung,\n",
|
|
" ganzzahlig: Sequence[str] = (),\n",
|
|
" toleranz: float = 1e-6) -> list[str]:\n",
|
|
" \"\"\"Prueft eine Loesung gegen die Anforderungen - OHNE den Solver zu fragen.\n",
|
|
"\n",
|
|
" Diese Funktion darf ausdruecklich KEIN Solverobjekt und keine\n",
|
|
" Modellvariable benutzen. Eine Pruefung aus denselben Bausteinen wie das\n",
|
|
" Modell prueft das Modell gegen sich selbst und findet nichts (siehe die\n",
|
|
" Denkfehler in den Kapiteln Graphen und MILP).\n",
|
|
"\n",
|
|
" Liefert eine Liste von Beanstandungen; leer heisst bestanden.\n",
|
|
" \"\"\"\n",
|
|
" beanstandungen: list[str] = []\n",
|
|
"\n",
|
|
" if not loesung.status.brauchbar:\n",
|
|
" return [f\"Kein verwertbares Ergebnis (Status: {loesung.status.value})\"]\n",
|
|
"\n",
|
|
" fehlend = [p.name for p in problem.produkte if p.name not in loesung.werte]\n",
|
|
" if fehlend:\n",
|
|
" return [f\"Loesung enthaelt keine Werte fuer: {fehlend}\"]\n",
|
|
"\n",
|
|
" mengen = np.array([loesung.werte[p.name] for p in problem.produkte])\n",
|
|
"\n",
|
|
" if (mengen < -toleranz).any():\n",
|
|
" negativ = [p.name for p, m in zip(problem.produkte, mengen) if m < -toleranz]\n",
|
|
" beanstandungen.append(f\"negative Mengen bei: {negativ}\")\n",
|
|
"\n",
|
|
" verbrauch = problem.verbrauchsmatrix() @ mengen\n",
|
|
" for ressource, ist, grenze in zip(problem.ressourcen, verbrauch,\n",
|
|
" problem.kapazitaetsvektor()):\n",
|
|
" if ist > grenze + toleranz:\n",
|
|
" beanstandungen.append(\n",
|
|
" f\"{ressource}: Verbrauch {ist:,.3f} ueber Kapazitaet {grenze:,.3f}\")\n",
|
|
"\n",
|
|
" for name in ganzzahlig:\n",
|
|
" wert = loesung.werte[name]\n",
|
|
" if abs(wert - round(wert)) > toleranz:\n",
|
|
" beanstandungen.append(f\"'{name}' = {wert!r} ist nicht ganzzahlig\")\n",
|
|
"\n",
|
|
" if loesung.zielwert is not None:\n",
|
|
" nachgerechnet = float(problem.deckungsbeitragsvektor() @ mengen)\n",
|
|
" if abs(nachgerechnet - loesung.zielwert) > toleranz * max(1.0, abs(nachgerechnet)):\n",
|
|
" beanstandungen.append(\n",
|
|
" f\"Zielwert {loesung.zielwert:,.4f} passt nicht zu den Mengen \"\n",
|
|
" f\"(nachgerechnet {nachgerechnet:,.4f})\")\n",
|
|
"\n",
|
|
" return beanstandungen\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 5. Ein Modellbauer je Solver - austauschbar ----------------------------\n",
|
|
"\n",
|
|
"def loese_mit_glop(problem: Produktionsproblem) -> Loesung:\n",
|
|
" \"\"\"Modellbauer fuer OR-Tools GLOP (kontinuierlich).\"\"\"\n",
|
|
" import time\n",
|
|
" from ortools.linear_solver import pywraplp\n",
|
|
"\n",
|
|
" solver = pywraplp.Solver.CreateSolver(\"GLOP\")\n",
|
|
" menge = {p.name: solver.NumVar(0, solver.infinity(), p.name)\n",
|
|
" for p in problem.produkte}\n",
|
|
" # Die Nebenbedingung wird gemerkt - ohne die Referenz gibt es spaeter\n",
|
|
" # keinen Schattenpreis (siehe Kapitel LP).\n",
|
|
" bedingung = {}\n",
|
|
" for ressource in problem.ressourcen:\n",
|
|
" bedingung[ressource] = solver.Add(\n",
|
|
" sum(menge[p.name] * p.verbrauch.get(ressource, 0.0)\n",
|
|
" for p in problem.produkte)\n",
|
|
" <= problem.kapazitaeten[ressource], name=ressource)\n",
|
|
" solver.Maximize(sum(menge[p.name] * p.deckungsbeitrag for p in problem.produkte))\n",
|
|
"\n",
|
|
" t0 = time.perf_counter()\n",
|
|
" rohstatus = solver.Solve()\n",
|
|
" laufzeit = time.perf_counter() - t0\n",
|
|
"\n",
|
|
" status = status_von_pywraplp(rohstatus)\n",
|
|
" if not status.brauchbar:\n",
|
|
" return Loesung(status=status, laufzeit=laufzeit)\n",
|
|
" return Loesung(status=status,\n",
|
|
" werte={name: v.solution_value() for name, v in menge.items()},\n",
|
|
" zielwert=solver.Objective().Value(),\n",
|
|
" schranke=solver.Objective().Value(),\n",
|
|
" schattenpreise={r: b.dual_value() for r, b in bedingung.items()},\n",
|
|
" laufzeit=laufzeit)\n",
|
|
"\n",
|
|
"\n",
|
|
"def loese_mit_scipy(problem: Produktionsproblem) -> Loesung:\n",
|
|
" \"\"\"Derselbe Fall mit scipy.optimize.linprog - anderer Solver, gleiche Loesung.\n",
|
|
"\n",
|
|
" Beachten Sie, wie wenig hier steht: Die Daten kommen fertig geprueft aus\n",
|
|
" dem Domaenenmodell, die Auswertung geht ans gemeinsame Loesung-Objekt.\n",
|
|
" Genau das ist der Ertrag der Trennung.\n",
|
|
" \"\"\"\n",
|
|
" import time\n",
|
|
" from scipy.optimize import linprog\n",
|
|
"\n",
|
|
" t0 = time.perf_counter()\n",
|
|
" ergebnis = linprog(c=-problem.deckungsbeitragsvektor(), # linprog minimiert\n",
|
|
" A_ub=problem.verbrauchsmatrix(),\n",
|
|
" b_ub=problem.kapazitaetsvektor(),\n",
|
|
" bounds=(0, None), method=\"highs\")\n",
|
|
" laufzeit = time.perf_counter() - t0\n",
|
|
"\n",
|
|
" status = status_von_scipy(ergebnis)\n",
|
|
" if not status.brauchbar:\n",
|
|
" return Loesung(status=status, laufzeit=laufzeit)\n",
|
|
" # linprog rechnet minimierend mit negierten Kosten - die Dualwerte\n",
|
|
" # muessen entsprechend zurueckgedreht werden.\n",
|
|
" return Loesung(status=status,\n",
|
|
" werte={p.name: float(x)\n",
|
|
" for p, x in zip(problem.produkte, ergebnis.x)},\n",
|
|
" zielwert=float(-ergebnis.fun),\n",
|
|
" schranke=float(-ergebnis.fun),\n",
|
|
" schattenpreise={r: float(-m) for r, m in\n",
|
|
" zip(problem.ressourcen,\n",
|
|
" ergebnis.ineqlin.marginals)},\n",
|
|
" laufzeit=laufzeit)\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 6. Ein- und Ausgabe: Tabellen sind die Wirklichkeit ---------------------\n",
|
|
"#\n",
|
|
"# In den meisten Betrieben liegen die Zahlen in einer Tabellenkalkulation\n",
|
|
"# (Kapitel Einfuehrung). Diese beiden Funktionen sind die einzige Stelle im\n",
|
|
"# Modul, die das weiss - alles andere arbeitet mit dem Domaenenmodell.\n",
|
|
"#\n",
|
|
"# pandas und openpyxl werden bewusst erst hier importiert: Wer or_kern nur\n",
|
|
"# fuer Statusauswertung und Pruefung benutzt, soll sie nicht installieren\n",
|
|
"# muessen.\n",
|
|
"\n",
|
|
"STAMMSPALTEN = (\"Produkt\", \"Deckungsbeitrag\")\n",
|
|
"\n",
|
|
"\n",
|
|
"def lade_produktionsproblem(pfad: str, blatt_produkte: str = \"Produkte\",\n",
|
|
" blatt_kapazitaeten: str = \"Kapazitaeten\"\n",
|
|
" ) -> Produktionsproblem:\n",
|
|
" \"\"\"Liest eine Excel-Mappe und liefert ein geprueftes Domaenenmodell.\n",
|
|
"\n",
|
|
" Alle Spalten ausser 'Produkt' und 'Deckungsbeitrag' gelten als\n",
|
|
" Verbrauchsspalten - eine neue Ressource ist damit eine neue SPALTE in der\n",
|
|
" Mappe und erfordert keine Codeaenderung.\n",
|
|
"\n",
|
|
" Die eigentliche Pruefung passiert nicht hier, sondern im Konstruktor von\n",
|
|
" Produktionsproblem. Diese Funktion uebersetzt nur Tabelle in Objekt; was\n",
|
|
" ein gueltiges Problem ist, steht an genau einer Stelle.\n",
|
|
" \"\"\"\n",
|
|
" import pandas as pd\n",
|
|
"\n",
|
|
" produkte_tabelle = pd.read_excel(pfad, sheet_name=blatt_produkte)\n",
|
|
" kapazitaeten_tabelle = pd.read_excel(pfad, sheet_name=blatt_kapazitaeten)\n",
|
|
"\n",
|
|
" fehlende = [s for s in STAMMSPALTEN if s not in produkte_tabelle.columns]\n",
|
|
" if fehlende:\n",
|
|
" raise ValueError(f\"Blatt '{blatt_produkte}': Spalten fehlen: {fehlende}\")\n",
|
|
" if produkte_tabelle.isna().any().any():\n",
|
|
" zeilen = produkte_tabelle[produkte_tabelle.isna().any(axis=1)].index.tolist()\n",
|
|
" raise ValueError(f\"Blatt '{blatt_produkte}': leere Zellen in Zeile(n) {zeilen}\")\n",
|
|
"\n",
|
|
" ressourcen = [s for s in produkte_tabelle.columns if s not in STAMMSPALTEN]\n",
|
|
" produkte = [\n",
|
|
" Produkt(name=str(zeile.Produkt),\n",
|
|
" deckungsbeitrag=float(zeile.Deckungsbeitrag),\n",
|
|
" verbrauch={r: float(getattr(zeile, r)) for r in ressourcen})\n",
|
|
" for zeile in produkte_tabelle.itertuples()\n",
|
|
" ]\n",
|
|
" kapazitaeten = {str(z.Ressource): float(z.Verfuegbar)\n",
|
|
" for z in kapazitaeten_tabelle.itertuples()}\n",
|
|
"\n",
|
|
" return Produktionsproblem(produkte=produkte, kapazitaeten=kapazitaeten)\n",
|
|
"\n",
|
|
"\n",
|
|
"def schreibe_ergebnis(pfad: str, problem: Produktionsproblem,\n",
|
|
" loesung: Loesung) -> None:\n",
|
|
" \"\"\"Schreibt Plan und Kennzahlen als zwei Blaetter zurueck nach Excel.\n",
|
|
"\n",
|
|
" Die Fachabteilung bekommt das Ergebnis im Format, das sie ohnehin benutzt.\n",
|
|
" Akzeptanz entscheidet ueber Projekterfolg (Kapitel Praxisfallen).\n",
|
|
" \"\"\"\n",
|
|
" import pandas as pd\n",
|
|
"\n",
|
|
" if not loesung.status.brauchbar:\n",
|
|
" raise ValueError(f\"Nichts zu schreiben - Status: {loesung.status.value}\")\n",
|
|
"\n",
|
|
" mengen = np.array([loesung.werte[p.name] for p in problem.produkte])\n",
|
|
" plan = pd.DataFrame({\n",
|
|
" \"Produkt\": [p.name for p in problem.produkte],\n",
|
|
" \"Menge\": mengen,\n",
|
|
" \"Deckungsbeitrag\": mengen * problem.deckungsbeitragsvektor(),\n",
|
|
" })\n",
|
|
"\n",
|
|
" verbrauch = problem.verbrauchsmatrix() @ mengen\n",
|
|
" kennzahlen: dict[str, float | str] = {\n",
|
|
" \"Status\": loesung.status.value,\n",
|
|
" \"Gesamtdeckungsbeitrag\": loesung.zielwert or 0.0,\n",
|
|
" }\n",
|
|
" for ressource, ist, grenze in zip(problem.ressourcen, verbrauch,\n",
|
|
" problem.kapazitaetsvektor()):\n",
|
|
" kennzahlen[f\"{ressource}: Verbrauch\"] = float(ist)\n",
|
|
" kennzahlen[f\"{ressource}: Auslastung %\"] = float(100.0 * ist / grenze)\n",
|
|
" if ressource in loesung.schattenpreise:\n",
|
|
" kennzahlen[f\"{ressource}: Schattenpreis\"] = \\\n",
|
|
" loesung.schattenpreise[ressource]\n",
|
|
"\n",
|
|
" kennzahlen_tabelle = pd.DataFrame({\n",
|
|
" \"Kennzahl\": list(kennzahlen),\n",
|
|
" \"Wert\": [round(w, 2) if isinstance(w, float) else w\n",
|
|
" for w in kennzahlen.values()],\n",
|
|
" })\n",
|
|
"\n",
|
|
" with pd.ExcelWriter(pfad, engine=\"openpyxl\") as mappe:\n",
|
|
" plan.round(2).to_excel(mappe, sheet_name=\"Plan\", index=False)\n",
|
|
" kennzahlen_tabelle.to_excel(mappe, sheet_name=\"Kennzahlen\", index=False)\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" # Die Schreinerei aus Kapitel Einfuehrung - jetzt als Domaenenmodell.\n",
|
|
" schreinerei = Produktionsproblem(\n",
|
|
" produkte=[\n",
|
|
" Produkt(name=\"Tisch\", deckungsbeitrag=240.0,\n",
|
|
" verbrauch={\"Montagestunden\": 3.0, \"Plattenmaterial\": 6.0}),\n",
|
|
" Produkt(name=\"Stuhl\", deckungsbeitrag=60.0,\n",
|
|
" verbrauch={\"Montagestunden\": 1.0, \"Plattenmaterial\": 1.0}),\n",
|
|
" ],\n",
|
|
" kapazitaeten={\"Montagestunden\": 150.0, \"Plattenmaterial\": 240.0},\n",
|
|
" )\n",
|
|
"\n",
|
|
" print(\"=\" * 78)\n",
|
|
" print(\" DERSELBE FALL, ZWEI SOLVER, EIN AUSWERTUNGSCODE\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
"\n",
|
|
" for name, bauer in [(\"OR-Tools GLOP\", loese_mit_glop),\n",
|
|
" (\"scipy / HiGHS\", loese_mit_scipy)]:\n",
|
|
" loesung = bauer(schreinerei)\n",
|
|
" beanstandungen = pruefe_loesung(schreinerei, loesung)\n",
|
|
" print(f\"\\n{name}\")\n",
|
|
" print(f\" {loesung.als_bericht()}\")\n",
|
|
" for produkt, menge in loesung.werte.items():\n",
|
|
" print(f\" {produkt:<10} {menge:8.2f}\")\n",
|
|
" print(f\" Abnahmepruefung: \"\n",
|
|
" f\"{'bestanden' if not beanstandungen else beanstandungen}\")\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"=\" * 78)\n",
|
|
" print(\" WAS DIE PRUEFUNGEN ABFANGEN\")\n",
|
|
" print(\"=\" * 78)\n",
|
|
"\n",
|
|
" faelle = [\n",
|
|
" (\"Kapazitaet 0\", dict(\n",
|
|
" produkte=schreinerei.produkte,\n",
|
|
" kapazitaeten={\"Montagestunden\": 0.0, \"Plattenmaterial\": 240.0})),\n",
|
|
" (\"Ressource ohne Kapazitaet\", dict(\n",
|
|
" produkte=[Produkt(name=\"Regal\", deckungsbeitrag=130.0,\n",
|
|
" verbrauch={\"Lackieren\": 2.0})],\n",
|
|
" kapazitaeten={\"Montagestunden\": 150.0})),\n",
|
|
" (\"doppelter Produktname\", dict(\n",
|
|
" produkte=[schreinerei.produkte[0], schreinerei.produkte[0]],\n",
|
|
" kapazitaeten=schreinerei.kapazitaeten)),\n",
|
|
" ]\n",
|
|
" for beschreibung, daten in faelle:\n",
|
|
" try:\n",
|
|
" Produktionsproblem(**daten)\n",
|
|
" print(f\" {beschreibung:<28} NICHT erkannt (!)\")\n",
|
|
" except Exception as fehler:\n",
|
|
" meldung = str(fehler).splitlines()\n",
|
|
" kern = next((z.strip() for z in meldung\n",
|
|
" if \"Value error\" in z or \"greater than\" in z), meldung[-1])\n",
|
|
" print(f\" {beschreibung:<28} abgefangen: {kern[:44]}\")\n",
|
|
"\n",
|
|
" print(\"\\nAlle drei scheitern beim EINLESEN - nicht erst beim Loesen und\")\n",
|
|
" print(\"schon gar nicht erst im Bericht. Das ist der ganze Zweck der\")\n",
|
|
" print(\"Domaenenschicht.\")\n",
|
|
" print(\"=\" * 78)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Der Solverwechsel in der Praxis\n",
|
|
"\n",
|
|
"`Solverwechsel_CPSAT_HiGHS.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Solverwechsel_CPSAT_HiGHS.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Praxisfallen: Denselben Fall einmal mit CP-SAT und einmal mit HiGHS rechnen.\n",
|
|
"\n",
|
|
"Die Trennung aus or_kern.py behauptet, ein Solverwechsel koste genau EINEN\n",
|
|
"Baustein. Dieses Programm loest diese Behauptung ein. Aufgabe ist eine\n",
|
|
"Standortplanung: Welche Lager oeffnen wir, und wer beliefert welchen Kunden?\n",
|
|
"\n",
|
|
" Standortproblem (Pydantic, geprueft) <- gemeinsam\n",
|
|
" baue_und_loese_mit_cpsat() <- der EINE Baustein\n",
|
|
" baue_und_loese_mit_highs() <- ... in zwei Ausfuehrungen\n",
|
|
" Loesung (DTO) <- gemeinsam\n",
|
|
" pruefe_zuordnung() <- gemeinsam\n",
|
|
" berichte() <- gemeinsam\n",
|
|
"\n",
|
|
"Die beiden Modellbauer sind 39 und 52 Zeilen lang - sie sind der gesamte\n",
|
|
"solverabhaengige Teil des Programms. Alles andere wird zweimal benutzt und\n",
|
|
"einmal geschrieben.\n",
|
|
"\n",
|
|
"WARUM ZWEI PROZESSE? ortools und highspy bringen beide eine eigene\n",
|
|
"HiGHS-Kopie mit und lassen sich auf vielen Systemen nicht gemeinsam\n",
|
|
"importieren (Kapitel Oekosystem). Das Hauptprogramm startet deshalb fuer jeden\n",
|
|
"Solver einen eigenen Python-Prozess und laesst sich die Loesung als JSON\n",
|
|
"zurueckgeben - das DTO ist nicht nur eine Sprachregelung, sondern ein\n",
|
|
"Datenformat, das eine Prozessgrenze ueberlebt.\n",
|
|
"\n",
|
|
"Aufruf:\n",
|
|
" python3 Solverwechsel_CPSAT_HiGHS.py # beide, mit Vergleich\n",
|
|
" python3 Solverwechsel_CPSAT_HiGHS.py cpsat # nur der Kindprozess\n",
|
|
" python3 Solverwechsel_CPSAT_HiGHS.py highs\n",
|
|
"\n",
|
|
"Benoetigt: numpy, pydantic, ortools, highspy (jeweils im eigenen Prozess)\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from __future__ import annotations\n",
|
|
"\n",
|
|
"import multiprocessing\n",
|
|
"import time\n",
|
|
"from concurrent.futures import ProcessPoolExecutor\n",
|
|
"\n",
|
|
"import numpy as np\n",
|
|
"from pydantic import BaseModel, Field, model_validator\n",
|
|
"\n",
|
|
"from or_kern import Loesung, SolverStatus, status_von_cpsat, status_von_highs\n",
|
|
"\n",
|
|
"ZEITLIMIT = 30.0\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 1. Domaenenmodell: dieselben Fakten fuer beide Solver -------------------\n",
|
|
"\n",
|
|
"class Standortproblem(BaseModel):\n",
|
|
" \"\"\"Kapazitierte Standortplanung mit Einzelbelieferung.\n",
|
|
"\n",
|
|
" Nach dem Muster von Produktionsproblem in or_kern.py: Die Pruefungen\n",
|
|
" stehen im Konstruktor, nicht im Solvercode - sie gelten damit fuer beide\n",
|
|
" Solver, und sie schlagen beim Einlesen zu.\n",
|
|
"\n",
|
|
" Alle Kosten sind ganzzahlig (Euro). Das ist keine Bequemlichkeit, sondern\n",
|
|
" Voraussetzung: CP-SAT rechnet ausschliesslich ganzzahlig.\n",
|
|
" \"\"\"\n",
|
|
" lager: list[str] = Field(min_length=1)\n",
|
|
" kunden: list[str] = Field(min_length=1)\n",
|
|
" fixkosten: list[int] # je Lager, faellt bei Eroeffnung an\n",
|
|
" kapazitaet: list[int] # je Lager, in Paletten\n",
|
|
" bedarf: list[int] # je Kunde, in Paletten\n",
|
|
" transport: list[list[int]] # [Lager][Kunde], Kosten der Belieferung\n",
|
|
"\n",
|
|
" @model_validator(mode=\"after\")\n",
|
|
" def pruefe_masse(self) -> \"Standortproblem\":\n",
|
|
" n, m = len(self.lager), len(self.kunden)\n",
|
|
" if len(self.fixkosten) != n or len(self.kapazitaet) != n:\n",
|
|
" raise ValueError(f\"fixkosten/kapazitaet muessen {n} Eintraege haben\")\n",
|
|
" if len(self.bedarf) != m:\n",
|
|
" raise ValueError(f\"bedarf muss {m} Eintraege haben\")\n",
|
|
" if len(self.transport) != n or any(len(z) != m for z in self.transport):\n",
|
|
" raise ValueError(f\"transport muss {n} x {m} sein\")\n",
|
|
" if sum(self.kapazitaet) < sum(self.bedarf):\n",
|
|
" raise ValueError(f\"Gesamtkapazitaet {sum(self.kapazitaet)} deckt den \"\n",
|
|
" f\"Gesamtbedarf {sum(self.bedarf)} nicht\")\n",
|
|
" return self\n",
|
|
"\n",
|
|
" def schluessel(self, i: int, j: int) -> str:\n",
|
|
" \"\"\"Variablenname im Loesung-DTO - beide Modellbauer benutzen ihn.\"\"\"\n",
|
|
" return f\"{self.lager[i]}->{self.kunden[j]}\"\n",
|
|
"\n",
|
|
"\n",
|
|
"def beispielproblem(saat: int = 11) -> Standortproblem:\n",
|
|
" \"\"\"Sechs moegliche Lager, zwoelf Kunden - klein genug fuer beide Solver.\"\"\"\n",
|
|
" rng = np.random.default_rng(saat)\n",
|
|
" lager = [f\"Lager_{k}\" for k in \"ABCDEF\"]\n",
|
|
" kunden = [f\"Kunde_{k:02d}\" for k in range(1, 13)]\n",
|
|
" bedarf = rng.integers(10, 60, len(kunden))\n",
|
|
" return Standortproblem(\n",
|
|
" lager=lager,\n",
|
|
" kunden=kunden,\n",
|
|
" fixkosten=rng.integers(3000, 9000, len(lager)).tolist(),\n",
|
|
" kapazitaet=[int(bedarf.sum() * 0.45)] * len(lager),\n",
|
|
" bedarf=bedarf.tolist(),\n",
|
|
" transport=rng.integers(200, 1800, (len(lager), len(kunden))).tolist(),\n",
|
|
" )\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 2. Der eine Baustein, der sich aendert: der Modellbauer -----------------\n",
|
|
"\n",
|
|
"def baue_und_loese_mit_cpsat(problem: Standortproblem) -> Loesung:\n",
|
|
" \"\"\"CP-SAT: Bool-Variablen, ganzzahlige Koeffizienten, Minimize.\"\"\"\n",
|
|
" from ortools.sat.python import cp_model\n",
|
|
"\n",
|
|
" n, m = len(problem.lager), len(problem.kunden)\n",
|
|
" modell = cp_model.CpModel()\n",
|
|
" y = [modell.NewBoolVar(f\"offen_{i}\") for i in range(n)]\n",
|
|
" x = {(i, j): modell.NewBoolVar(f\"liefert_{i}_{j}\")\n",
|
|
" for i in range(n) for j in range(m)}\n",
|
|
"\n",
|
|
" for j in range(m): # jeder Kunde genau einmal\n",
|
|
" modell.AddExactlyOne(x[i, j] for i in range(n))\n",
|
|
" for i in range(n):\n",
|
|
" for j in range(m): # nur aus offenen Lagern\n",
|
|
" modell.AddImplication(x[i, j], y[i])\n",
|
|
" modell.Add(sum(problem.bedarf[j] * x[i, j] for j in range(m))\n",
|
|
" <= problem.kapazitaet[i] * y[i]) # Kapazitaet\n",
|
|
"\n",
|
|
" modell.Minimize(\n",
|
|
" sum(problem.fixkosten[i] * y[i] for i in range(n))\n",
|
|
" + sum(problem.transport[i][j] * x[i, j] for i in range(n) for j in range(m)))\n",
|
|
"\n",
|
|
" loeser = cp_model.CpSolver()\n",
|
|
" loeser.parameters.max_time_in_seconds = ZEITLIMIT\n",
|
|
" loeser.parameters.num_workers = 1\n",
|
|
" loeser.parameters.random_seed = 1\n",
|
|
" status = status_von_cpsat(loeser.Solve(modell))\n",
|
|
" if not status.brauchbar:\n",
|
|
" return Loesung(status=status, laufzeit=loeser.WallTime())\n",
|
|
"\n",
|
|
" return Loesung(\n",
|
|
" status=status,\n",
|
|
" werte={problem.schluessel(i, j): float(loeser.Value(x[i, j]))\n",
|
|
" for i in range(n) for j in range(m)},\n",
|
|
" zielwert=loeser.ObjectiveValue(),\n",
|
|
" schranke=loeser.BestObjectiveBound(),\n",
|
|
" laufzeit=loeser.WallTime())\n",
|
|
"\n",
|
|
"\n",
|
|
"def baue_und_loese_mit_highs(problem: Standortproblem) -> Loesung:\n",
|
|
" \"\"\"HiGHS: dieselben Restriktionen als Ungleichungszeilen einer Matrix.\"\"\"\n",
|
|
" import highspy\n",
|
|
"\n",
|
|
" n, m = len(problem.lager), len(problem.kunden)\n",
|
|
" anzahl_x = n * m # Spalten 0..n*m-1\n",
|
|
" spalte_y = lambda i: anzahl_x + i # danach die y_i # noqa: E731\n",
|
|
"\n",
|
|
" modell = highspy.Highs()\n",
|
|
" modell.setOptionValue(\"output_flag\", False)\n",
|
|
" modell.setOptionValue(\"time_limit\", ZEITLIMIT)\n",
|
|
" modell.addVars(anzahl_x + n, np.zeros(anzahl_x + n), np.ones(anzahl_x + n))\n",
|
|
" for spalte in range(anzahl_x + n):\n",
|
|
" modell.changeColIntegrality(spalte, highspy.HighsVarType.kInteger)\n",
|
|
" for i in range(n):\n",
|
|
" modell.changeColCost(spalte_y(i), float(problem.fixkosten[i]))\n",
|
|
" for j in range(m):\n",
|
|
" modell.changeColCost(i * m + j, float(problem.transport[i][j]))\n",
|
|
"\n",
|
|
" for j in range(m): # jeder Kunde genau einmal\n",
|
|
" index = np.array([i * m + j for i in range(n)], dtype=np.int32)\n",
|
|
" modell.addRow(1.0, 1.0, n, index, np.ones(n))\n",
|
|
" for i in range(n):\n",
|
|
" for j in range(m): # x_ij - y_i <= 0\n",
|
|
" modell.addRow(-highspy.kHighsInf, 0.0, 2,\n",
|
|
" np.array([i * m + j, spalte_y(i)], dtype=np.int32),\n",
|
|
" np.array([1.0, -1.0]))\n",
|
|
" index = np.array([i * m + j for j in range(m)] + [spalte_y(i)],\n",
|
|
" dtype=np.int32) # Kapazitaet\n",
|
|
" werte = np.concatenate([np.array(problem.bedarf, dtype=float),\n",
|
|
" [-float(problem.kapazitaet[i])]])\n",
|
|
" modell.addRow(-highspy.kHighsInf, 0.0, m + 1, index, werte)\n",
|
|
"\n",
|
|
" t0 = time.perf_counter()\n",
|
|
" modell.run()\n",
|
|
" laufzeit = time.perf_counter() - t0\n",
|
|
"\n",
|
|
" status = status_von_highs(modell.modelStatusToString(modell.getModelStatus()))\n",
|
|
" if not status.brauchbar:\n",
|
|
" return Loesung(status=status, laufzeit=laufzeit)\n",
|
|
"\n",
|
|
" loesungswerte = modell.getSolution().col_value\n",
|
|
" info = modell.getInfo()\n",
|
|
" return Loesung(\n",
|
|
" status=status,\n",
|
|
" werte={problem.schluessel(i, j): float(loesungswerte[i * m + j])\n",
|
|
" for i in range(n) for j in range(m)},\n",
|
|
" zielwert=info.objective_function_value,\n",
|
|
" schranke=info.mip_dual_bound,\n",
|
|
" laufzeit=laufzeit)\n",
|
|
"\n",
|
|
"\n",
|
|
"MODELLBAUER = {\"cpsat\": baue_und_loese_mit_cpsat, \"highs\": baue_und_loese_mit_highs}\n",
|
|
"\n",
|
|
"\n",
|
|
"# --- 3. Alles Weitere ist wieder gemeinsam -----------------------------------\n",
|
|
"\n",
|
|
"def pruefe_zuordnung(problem: Standortproblem, loesung: Loesung,\n",
|
|
" toleranz: float = 1e-6) -> list[str]:\n",
|
|
" \"\"\"Prueft die Loesung gegen die Anforderung - ohne Solver, ohne Modell.\"\"\"\n",
|
|
" if not loesung.status.brauchbar:\n",
|
|
" return [f\"kein verwertbares Ergebnis ({loesung.status.value})\"]\n",
|
|
"\n",
|
|
" n, m = len(problem.lager), len(problem.kunden)\n",
|
|
" zuordnung = np.array([[loesung.werte[problem.schluessel(i, j)]\n",
|
|
" for j in range(m)] for i in range(n)])\n",
|
|
" beanstandungen: list[str] = []\n",
|
|
"\n",
|
|
" if (np.abs(zuordnung - np.round(zuordnung)) > toleranz).any():\n",
|
|
" beanstandungen.append(\"Zuordnungen sind nicht 0/1\")\n",
|
|
" zuordnung = np.round(zuordnung)\n",
|
|
"\n",
|
|
" for j, kunde in enumerate(problem.kunden):\n",
|
|
" if abs(zuordnung[:, j].sum() - 1.0) > toleranz:\n",
|
|
" beanstandungen.append(f\"{kunde} wird {zuordnung[:, j].sum():.0f}-mal beliefert\")\n",
|
|
"\n",
|
|
" beliefert = zuordnung @ np.array(problem.bedarf, dtype=float)\n",
|
|
" for i, lagername in enumerate(problem.lager):\n",
|
|
" if beliefert[i] > problem.kapazitaet[i] + toleranz:\n",
|
|
" beanstandungen.append(f\"{lagername}: {beliefert[i]:.0f} Paletten ueber \"\n",
|
|
" f\"Kapazitaet {problem.kapazitaet[i]}\")\n",
|
|
"\n",
|
|
" if loesung.zielwert is not None:\n",
|
|
" offen = beliefert > toleranz\n",
|
|
" nachgerechnet = (np.array(problem.fixkosten, dtype=float) @ offen\n",
|
|
" + (np.array(problem.transport, dtype=float) * zuordnung).sum())\n",
|
|
" if abs(nachgerechnet - loesung.zielwert) > 0.5:\n",
|
|
" beanstandungen.append(f\"Zielwert {loesung.zielwert:,.0f} passt nicht zur \"\n",
|
|
" f\"Zuordnung (nachgerechnet {nachgerechnet:,.0f})\")\n",
|
|
" return beanstandungen\n",
|
|
"\n",
|
|
"\n",
|
|
"def geoeffnete_lager(problem: Standortproblem, loesung: Loesung) -> list[str]:\n",
|
|
" m = len(problem.kunden)\n",
|
|
" return [name for i, name in enumerate(problem.lager)\n",
|
|
" if any(loesung.werte[problem.schluessel(i, j)] > 0.5 for j in range(m))]\n",
|
|
"\n",
|
|
"\n",
|
|
"def loese_in_eigenem_prozess(name: str, problem: Standortproblem) -> Loesung:\n",
|
|
" \"\"\"Laesst genau einen Modellbauer in einem frischen Prozess rechnen.\n",
|
|
"\n",
|
|
" 'spawn' statt des Linux-Standards 'fork': Der Kindprozess startet mit\n",
|
|
" einem leeren Interpreter und importiert nur den Solver, den SEIN\n",
|
|
" Modellbauer braucht. max_tasks_per_child=1 sorgt dafuer, dass der Pool\n",
|
|
" seinen Arbeiter nicht wiederverwendet - sonst saessen beim zweiten Aufruf\n",
|
|
" wieder beide Bibliotheken im selben Prozess.\n",
|
|
"\n",
|
|
" Hin und zurueck wandert das Domaenenmodell bzw. das Loesungs-DTO. Beide\n",
|
|
" kennen keinen Solver, sind also serialisierbar - genau dafuer sind sie da.\n",
|
|
" \"\"\"\n",
|
|
" with ProcessPoolExecutor(\n",
|
|
" max_workers=1,\n",
|
|
" mp_context=multiprocessing.get_context(\"spawn\"),\n",
|
|
" max_tasks_per_child=1) as pool:\n",
|
|
" return pool.submit(MODELLBAUER[name], problem).result(timeout=300)\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" problem = beispielproblem()\n",
|
|
"\n",
|
|
" # --- Beide Solver anstossen und vergleichen ---------------------------\n",
|
|
" print(\"=\" * 82)\n",
|
|
" print(\" DERSELBE FALL, ZWEI SOLVER - UND EIN AUSWERTUNGSCODE\")\n",
|
|
" print(\"=\" * 82)\n",
|
|
" print(f\"Standortplanung: {len(problem.lager)} moegliche Lager, \"\n",
|
|
" f\"{len(problem.kunden)} Kunden, {sum(problem.bedarf)} Paletten Bedarf.\")\n",
|
|
" print(f\"Kapazitaet je Lager: {problem.kapazitaet[0]} Paletten \"\n",
|
|
" f\"-> mindestens 3 Lager noetig.\\n\")\n",
|
|
"\n",
|
|
" loesungen: dict[str, Loesung] = {}\n",
|
|
" for name, beschriftung in [(\"cpsat\", \"OR-Tools CP-SAT\"),\n",
|
|
" (\"highs\", \"HiGHS (highspy)\")]:\n",
|
|
" loesung = loesungen[name] = loese_in_eigenem_prozess(name, problem)\n",
|
|
" beanstandungen = pruefe_zuordnung(problem, loesung)\n",
|
|
"\n",
|
|
" print(f\"{beschriftung}\")\n",
|
|
" print(f\" {loesung.als_bericht()}\")\n",
|
|
" print(f\" eroeffnete Lager: {', '.join(geoeffnete_lager(problem, loesung))}\")\n",
|
|
" print(f\" Abnahmepruefung: \"\n",
|
|
" f\"{'bestanden' if not beanstandungen else beanstandungen}\")\n",
|
|
"\n",
|
|
" # --- Was der Vergleich zeigt -----------------------------------------\n",
|
|
" zielwerte = [loesung.zielwert for loesung in loesungen.values()]\n",
|
|
" print(\"-\" * 82)\n",
|
|
" print(f\"Zielwertdifferenz: {abs(zielwerte[0] - zielwerte[1]):.6f} EUR\")\n",
|
|
"\n",
|
|
" gleich_belegt = all(\n",
|
|
" round(loesungen[\"cpsat\"].werte[s]) == round(loesungen[\"highs\"].werte[s])\n",
|
|
" for s in loesungen[\"cpsat\"].werte)\n",
|
|
" print(f\"Identische Zuordnung: {'ja' if gleich_belegt else 'nein'}\")\n",
|
|
"\n",
|
|
" assert abs(zielwerte[0] - zielwerte[1]) < 0.5, \"Die Solver widersprechen sich!\"\n",
|
|
" assert all(l.status is SolverStatus.OPTIMAL for l in loesungen.values())\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"=\" * 82)\n",
|
|
" print(\" WAS DER WECHSEL GEKOSTET HAT\")\n",
|
|
" print(\"=\" * 82)\n",
|
|
" print(\"Ausgetauscht wurde EINE Funktion. Domaenenmodell, Abnahmepruefung und\")\n",
|
|
" print(\"Bericht sind woertlich dieselben - sie sehen den Solver nie.\")\n",
|
|
" print()\n",
|
|
" print(\"Nicht umsonst ist der Wechsel trotzdem:\")\n",
|
|
" print(\" * CP-SAT rechnet ausschliesslich GANZZAHLIG. Alle Kosten sind hier\")\n",
|
|
" print(\" deshalb int. Wer in Euro und Cent rechnet, skaliert vorher auf Cent -\")\n",
|
|
" print(\" und muss das im Bericht wieder zuruecknehmen.\")\n",
|
|
" print(\" * HiGHS braucht die Restriktionen als Matrixzeilen, CP-SAT nimmt sie\")\n",
|
|
" print(\" als Ausdruecke. Das ist der Grund, warum der HiGHS-Modellbauer\")\n",
|
|
" print(\" laenger ist, obwohl er dasselbe Modell beschreibt.\")\n",
|
|
" print(\" * Beide Bibliotheken bringen eine eigene HiGHS-Kopie mit und lassen\")\n",
|
|
" print(\" sich nicht gemeinsam importieren - daher die zwei Prozesse.\")\n",
|
|
" print()\n",
|
|
" print(\"Der Ertrag: Beide beweisen denselben optimalen Zielwert, und die\")\n",
|
|
" print(\"Entscheidung zwischen ihnen ist eine Frage der Laufzeit geworden -\")\n",
|
|
" print(\"nicht eine Frage, wie viel Code man neu schreiben muss.\")\n",
|
|
" print()\n",
|
|
" print(\"Verglichen wird deshalb der ZIELWERT, nicht der Plan: Gibt es mehrere\")\n",
|
|
" print(\"gleich teure Loesungen, darf jeder Solver eine andere davon liefern.\")\n",
|
|
" print(\"Hier stimmen sie zufaellig ueberein - darauf zu testen waere trotzdem\")\n",
|
|
" print(\"ein unzuverlaessiger Test (siehe JobShop_Intervalle.py).\")\n",
|
|
" print(\"=\" * 82)"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "markdown",
|
|
"metadata": {},
|
|
"source": [
|
|
"## Finde den Denkfehler\n",
|
|
"\n",
|
|
"`Betriebsueberwachung.py`\n"
|
|
]
|
|
},
|
|
{
|
|
"cell_type": "code",
|
|
"execution_count": null,
|
|
"metadata": {},
|
|
"outputs": [],
|
|
"source": [
|
|
"#!/usr/bin/env python3\n",
|
|
"\n",
|
|
"# Betriebsueberwachung.py\n",
|
|
"\"\"\"\n",
|
|
"Kapitel Praxisfallen: Das Modell, das still aufgehoert hat zu optimieren.\n",
|
|
"\n",
|
|
"Ein Optimierungsmodell geht in Betrieb und laeuft jede Nacht. Es liefert\n",
|
|
"zuverlaessig eine Zahl, die Job-Dauer bleibt konstant, es gibt keine\n",
|
|
"Fehlermeldung. Alles sieht gut aus.\n",
|
|
"\n",
|
|
"Waehrenddessen waechst das Unternehmen. Aus 15 Lagern werden 40, dann 80. Das\n",
|
|
"Zeitlimit von drei Sekunden - vor zwei Jahren grosszuegig bemessen - reicht\n",
|
|
"irgendwann nicht mehr. Der Solver bricht ab und liefert die beste Loesung, die\n",
|
|
"er bis dahin gefunden hat.\n",
|
|
"\n",
|
|
"Das ist nicht falsch. Es ist sogar richtig so. Aber es ist etwas anderes als\n",
|
|
"das, was das Modell einmal geleistet hat - und niemand merkt es, weil im\n",
|
|
"Protokoll nur die Kosten stehen.\n",
|
|
"\n",
|
|
"Dieses Programm zeigt die Verschlechterung und die Ueberwachung, die sie\n",
|
|
"sichtbar macht.\n",
|
|
"\n",
|
|
"WICHTIG: ortools wird hier nicht importiert (Konflikt mit highspy, siehe\n",
|
|
"Kapitel Oekosystem).\n",
|
|
"\n",
|
|
"Benoetigt: numpy, highspy\n",
|
|
"\"\"\"\n",
|
|
"\n",
|
|
"from __future__ import annotations\n",
|
|
"\n",
|
|
"import time\n",
|
|
"from dataclasses import dataclass\n",
|
|
"\n",
|
|
"import numpy as np\n",
|
|
"import highspy\n",
|
|
"\n",
|
|
"ZEITLIMIT = 3.0 # das Budget, das der Nachtjob hat\n",
|
|
"GAP_WARNSCHWELLE = 0.02 # ab 2 % Gap wollen wir es wissen\n",
|
|
"ZEITAUSLASTUNG_WARNSCHWELLE = 0.9\n",
|
|
"\n",
|
|
"\n",
|
|
"@dataclass\n",
|
|
"class Laufprotokoll:\n",
|
|
" \"\"\"Was nach JEDEM Produktivlauf protokolliert gehoert.\n",
|
|
"\n",
|
|
" Die meisten Systeme schreiben nur 'kosten'. Genau deshalb faellt eine\n",
|
|
" schleichende Verschlechterung ueber Monate nicht auf.\n",
|
|
" \"\"\"\n",
|
|
" zeitpunkt: str\n",
|
|
" lager: int\n",
|
|
" kunden: int\n",
|
|
" status: str\n",
|
|
" kosten: float\n",
|
|
" gap: float\n",
|
|
" laufzeit: float\n",
|
|
" knoten: int\n",
|
|
"\n",
|
|
" @property\n",
|
|
" def warnungen(self) -> list[str]:\n",
|
|
" meldungen = []\n",
|
|
" if self.status != \"Optimal\":\n",
|
|
" meldungen.append(f\"nicht beweisbar optimal ({self.status})\")\n",
|
|
" if self.gap > GAP_WARNSCHWELLE:\n",
|
|
" meldungen.append(f\"Gap {self.gap:.1%} ueber Schwelle \"\n",
|
|
" f\"{GAP_WARNSCHWELLE:.0%}\")\n",
|
|
" if self.laufzeit > ZEITAUSLASTUNG_WARNSCHWELLE * ZEITLIMIT:\n",
|
|
" meldungen.append(f\"Zeitbudget zu {self.laufzeit / ZEITLIMIT:.0%} \"\n",
|
|
" f\"ausgeschoepft\")\n",
|
|
" return meldungen\n",
|
|
"\n",
|
|
"\n",
|
|
"def plane_netzwerk(n_lager: int, n_kunden: int, zeitlimit: float,\n",
|
|
" saat: int = 7) -> tuple[str, float, float, float, int]:\n",
|
|
" \"\"\"Standortplanung wie in Kapitel MILP, nur mit wachsender Groesse.\"\"\"\n",
|
|
" rng = np.random.default_rng(saat)\n",
|
|
" fixkosten = rng.uniform(3000, 9000, n_lager)\n",
|
|
" transport = rng.uniform(5, 60, (n_lager, n_kunden))\n",
|
|
" bedarf = rng.uniform(10, 60, n_kunden)\n",
|
|
" kapazitaet = np.full(n_lager, bedarf.sum() * 0.22)\n",
|
|
"\n",
|
|
" modell = highspy.Highs()\n",
|
|
" modell.setOptionValue(\"output_flag\", False)\n",
|
|
" modell.setOptionValue(\"time_limit\", zeitlimit)\n",
|
|
"\n",
|
|
" anzahl_x = n_lager * n_kunden\n",
|
|
" unendlich = highspy.kHighsInf\n",
|
|
" modell.addVars(anzahl_x, np.zeros(anzahl_x), np.full(anzahl_x, unendlich))\n",
|
|
" modell.addVars(n_lager, np.zeros(n_lager), np.ones(n_lager))\n",
|
|
" for i in range(n_lager):\n",
|
|
" modell.changeColIntegrality(anzahl_x + i, highspy.HighsVarType.kInteger)\n",
|
|
" modell.changeColCost(anzahl_x + i, fixkosten[i])\n",
|
|
" for j in range(n_kunden):\n",
|
|
" modell.changeColCost(i * n_kunden + j, transport[i, j])\n",
|
|
"\n",
|
|
" for j in range(n_kunden):\n",
|
|
" index = np.array([i * n_kunden + j for i in range(n_lager)], dtype=np.int32)\n",
|
|
" modell.addRow(bedarf[j], bedarf[j], len(index), index, np.ones(len(index)))\n",
|
|
" for i in range(n_lager):\n",
|
|
" index = np.array([i * n_kunden + j for j in range(n_kunden)] + [anzahl_x + i],\n",
|
|
" dtype=np.int32)\n",
|
|
" werte = np.concatenate([np.ones(n_kunden), [-kapazitaet[i]]])\n",
|
|
" modell.addRow(-unendlich, 0.0, len(index), index, werte)\n",
|
|
"\n",
|
|
" t0 = time.perf_counter()\n",
|
|
" modell.run()\n",
|
|
" laufzeit = time.perf_counter() - t0\n",
|
|
" info = modell.getInfo()\n",
|
|
" return (modell.modelStatusToString(modell.getModelStatus()),\n",
|
|
" info.objective_function_value, info.mip_gap, laufzeit,\n",
|
|
" info.mip_node_count)\n",
|
|
"\n",
|
|
"\n",
|
|
"if __name__ == \"__main__\":\n",
|
|
" # So ist das Unternehmen ueber zwei Jahre gewachsen.\n",
|
|
" entwicklung = [(\"2024-Q1\", 15, 40), (\"2024-Q3\", 25, 70),\n",
|
|
" (\"2025-Q1\", 40, 110), (\"2025-Q3\", 60, 160),\n",
|
|
" (\"2026-Q1\", 80, 220)]\n",
|
|
"\n",
|
|
" print(\"=\" * 84)\n",
|
|
" print(\" WAS DAS PROTOKOLL ZEIGT - UND WAS ES ZEIGEN SOLLTE\")\n",
|
|
" print(\"=\" * 84)\n",
|
|
" print(f\"Derselbe Nachtjob, unveraendert, mit {ZEITLIMIT:.0f} Sekunden Zeitlimit.\\n\")\n",
|
|
"\n",
|
|
" protokolle = []\n",
|
|
" for zeitpunkt, lager, kunden in entwicklung:\n",
|
|
" status, kosten, gap, laufzeit, knoten = plane_netzwerk(\n",
|
|
" lager, kunden, ZEITLIMIT)\n",
|
|
" protokolle.append(Laufprotokoll(zeitpunkt, lager, kunden, status,\n",
|
|
" kosten, gap, laufzeit, knoten))\n",
|
|
"\n",
|
|
" print(\"So sieht das ueblich gefuehrte Protokoll aus:\\n\")\n",
|
|
" print(f\" {'Zeitpunkt':<10} {'Kosten':>12}\")\n",
|
|
" print(\" \" + \"-\" * 24)\n",
|
|
" for p in protokolle:\n",
|
|
" print(f\" {p.zeitpunkt:<10} {p.kosten:>12,.0f}\")\n",
|
|
" print(\"\\n -> Die Kosten steigen. Das Unternehmen waechst ja auch.\")\n",
|
|
" print(\" Nichts an dieser Tabelle deutet auf ein Problem hin.\")\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"-\" * 84)\n",
|
|
" print(\"So sieht ein vollstaendiges Protokoll aus:\\n\")\n",
|
|
" print(f\" {'Zeitpunkt':<10} {'Groesse':>10} {'Status':>20} {'Kosten':>11} \"\n",
|
|
" f\"{'Gap':>8} {'Zeit':>7}\")\n",
|
|
" print(\" \" + \"-\" * 80)\n",
|
|
" for p in protokolle:\n",
|
|
" groesse = f\"{p.lager}x{p.kunden}\"\n",
|
|
" print(f\" {p.zeitpunkt:<10} {groesse:>10} {p.status:>20} \"\n",
|
|
" f\"{p.kosten:>11,.0f} {p.gap * 100:7.2f} % {p.laufzeit:6.2f}s\")\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"-\" * 84)\n",
|
|
" print(\"Und so sehen die Warnungen aus, die daraus folgen:\\n\")\n",
|
|
" for p in protokolle:\n",
|
|
" if p.warnungen:\n",
|
|
" print(f\" {p.zeitpunkt}: \" + \"; \".join(p.warnungen))\n",
|
|
" else:\n",
|
|
" print(f\" {p.zeitpunkt}: in Ordnung\")\n",
|
|
"\n",
|
|
" letzte = protokolle[-1]\n",
|
|
" erste_warnung = next(p for p in protokolle if p.warnungen)\n",
|
|
"\n",
|
|
" print(\"\\n\" + \"=\" * 84)\n",
|
|
" print(\" WAS DA PASSIERT IST\")\n",
|
|
" print(\"=\" * 84)\n",
|
|
" print(f\"Seit {erste_warnung.zeitpunkt} erreicht der Job das Zeitlimit und liefert\")\n",
|
|
" print(\"nicht mehr das Optimum, sondern die beste bis dahin gefundene Loesung.\")\n",
|
|
" print(f\"Im letzten Lauf betraegt der Abstand zum Bestmoeglichen \"\n",
|
|
" f\"{letzte.gap:.1%}.\")\n",
|
|
" print()\n",
|
|
" print(\"Der Job ist nicht abgestuerzt. Er hat keine Fehlermeldung erzeugt.\")\n",
|
|
" print(\"Die Laufzeit ist sogar bemerkenswert STABIL geblieben - genau deshalb,\")\n",
|
|
" print(\"weil das Zeitlimit greift. Ein Ueberwachungssystem, das auf Abstuerze\")\n",
|
|
" print(\"und Laufzeitspitzen achtet, sieht hier nichts.\")\n",
|
|
" print()\n",
|
|
" print(\"Drei Zahlen gehoeren deshalb in jedes Protokoll eines Optimierungsjobs:\")\n",
|
|
" print(\" * der STATUS - 'Optimal' oder etwas anderes?\")\n",
|
|
" print(\" * der GAP - wie weit ist die Loesung vom Bestmoeglichen entfernt?\")\n",
|
|
" print(\" * die LAUFZEIT im Verhaeltnis zum Limit - wie nah am Anschlag?\")\n",
|
|
" print()\n",
|
|
" print(\"Die dritte ist die frueheste Warnung: Sie steigt, lange bevor der Gap\")\n",
|
|
" print(\"sichtbar wird, und gibt Zeit zum Handeln.\")\n",
|
|
" print(\"=\" * 84)"
|
|
]
|
|
}
|
|
],
|
|
"metadata": {
|
|
"kernelspec": {
|
|
"display_name": "Python 3",
|
|
"language": "python",
|
|
"name": "python3"
|
|
},
|
|
"language_info": {
|
|
"name": "python",
|
|
"version": "3.11"
|
|
}
|
|
},
|
|
"nbformat": 4,
|
|
"nbformat_minor": 5
|
|
}
|