operations_research/OR_HTML_04/Notebooks_04/cpsat.ipynb
dschlueter 293370b7a1 Phase 6.3: Parallelitaet und Determinismus - gemessen statt behauptet
Neuer Abschnitt im CP-SAT-Kapitel plus Parallele_Suche.py (76. Programm).
Setzt Paket 2 aus Verbesserungen_02.md um.

Was fehlte, war nicht die Regel, sondern die Messung: Das Buch setzt an acht
Stellen num_workers = 1 mit dem Kommentar "fuer eine reproduzierbare
Ausgabe", nachgeprueft hatte es nie jemand. Gerechnet wird auf demselben
Job-Shop wie das Intervallvariablen-Kapitel, nur gross genug, dass die Suche
arbeitet: 12 Auftraege, 10 Maschinen, 120 Arbeitsgaenge.

Zwei Befunde, beide staerker als die Behauptung:

* Die Beschleunigung ist ueberlinear. Acht Arbeiter waren im abgedruckten
  Lauf nicht achtmal, sondern 12,3-mal schneller als einer. Kein Messfehler:
  CP-SAT vervielfacht nicht dieselbe Suche, sondern laesst verschiedene
  Strategien nebeneinander laufen, die einander ihre Schranken mitteilen.
* Der Seed genuegt nicht - und zwar schon ab ZWEI Arbeitern. Ein Arbeiter:
  1 Plan aus 4 Laeufen. Zwei Arbeiter: 3 verschiedene Plaene aus 4 Laeufen,
  bei identischem random_seed und identischem Zielwert 183.

Die vollstaendige Antwort kam erst ueber die Uebungsaufgabe: Mit
num_workers = 1, aber fuenf verschiedenen Seeds ergeben sich ebenfalls fuenf
verschiedene Plaene. Keiner der beiden Parameter sichert die
Reproduzierbarkeit allein - erst die Kombination traegt.

Weiter fuer die Aufgabe gemessen: Der Gewinn kehrt sich um (auf 24 Kernen
Faktor 10,2 bei 8 Arbeitern, 12,3 bei 16, 9,0 bei 24) - "so viele Arbeiter
wie Kerne" ist damit widerlegt. Und bei 15 Auftraegen laeuft ein Arbeiter
ins 60-s-Limit (FEASIBLE, Makespan 200), waehrend acht OPTIMAL mit
demselben Makespan 200 nach 26,4 s melden: Der Unterschied liegt nicht in
der Loesung, sondern im Beweis, dass es keine bessere gibt.

Die abgedruckte Ausgabe traegt die Kennzeichnung "Laufzeiten und die Zahl
der verschiedenen Plaene sind hardwareabhaengig" - nach dem Muster, das das
Testing-Kapitel fuer Benchmark_Skalierung.py schon verwendet. Der Vergleich
des extrahierten Programms mit dem Abdruck weicht denn auch in genau einer
Zelle ab (4 statt 3 verschiedene Plaene bei 4 Arbeitern); Zielwert und
Struktur sind identisch. Hier ist die Nichtreproduzierbarkeit der
abgedruckten Zahl die Aussage selbst.

Mitgezogen: Kapitelkopf, Lernziele, Selbsttest, Zusammenfassung,
Vorwort-Programmverzeichnis, Uebungsaufgabe und Loesung in Anhang A, ein
Verweis aus dem bestehenden Callout zu mehrdeutigen Optima und einer aus
Warmstart_Effekt.py im MILP-Kapitel (dort nach Regel 12 der Kapitelname).

Stand: 295 Abschnitte, 730 Querverweise, 328 Indexmarken, 76 Programme,
140 Aufgaben mit 140 Loesungen, 33 pytest-Tests, PDF 744 Seiten, 69
netzfreie Programme fehlerfrei.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-09-08 11:19:53 +02:00

1114 lines
52 KiB
Text
Generated

{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# Kapitel 7: Constraint Programming mit CP-SAT — Logik, Scheduling und Zuweisung\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": [
"## Was Propagation leistet\n",
"\n",
"`Propagation_Demo.py`\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#!/usr/bin/env python3\n",
"\n",
"# Propagation_Demo.py\n",
"\"\"\"\n",
"Kapitel CP-SAT: Propagation sichtbar machen.\n",
"Vergleicht die Zahl der geprueften Kombinationen bei roher Aufzaehlung\n",
"mit der Zahl der Verzweigungen, die CP-SAT tatsaechlich braucht.\n",
"\"\"\"\n",
"\n",
"import itertools\n",
"\n",
"from ortools.sat.python import cp_model\n",
"\n",
"\n",
"def rohe_aufzaehlung(n: int, obergrenze: int) -> tuple[int, int]:\n",
" \"\"\"Zaehlt ALLE Kombinationen und prueft jede einzeln (Brute Force).\"\"\"\n",
" geprueft = 0\n",
" loesungen = 0\n",
" for kombination in itertools.product(range(1, obergrenze + 1), repeat=n):\n",
" geprueft += 1\n",
" if len(set(kombination)) == n and sum(kombination) == 3 * n:\n",
" loesungen += 1\n",
" return geprueft, loesungen\n",
"\n",
"\n",
"def mit_cp_sat(n: int, obergrenze: int) -> tuple[int, int, float]:\n",
" \"\"\"Dasselbe Problem deklarativ: alle verschieden, Summe = 3n.\"\"\"\n",
" modell = cp_model.CpModel()\n",
" x = [modell.NewIntVar(1, obergrenze, f\"x{i}\") for i in range(n)]\n",
" modell.AddAllDifferent(x) # globales Constraint\n",
" modell.Add(sum(x) == 3 * n)\n",
"\n",
" loeser = cp_model.CpSolver()\n",
" loeser.parameters.enumerate_all_solutions = True\n",
"\n",
" class Zaehler(cp_model.CpSolverSolutionCallback):\n",
" def __init__(self):\n",
" super().__init__()\n",
" self.anzahl = 0\n",
"\n",
" def on_solution_callback(self):\n",
" self.anzahl += 1\n",
"\n",
" zaehler = Zaehler()\n",
" loeser.Solve(modell, zaehler)\n",
" return zaehler.anzahl, loeser.NumBranches(), loeser.WallTime()\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" print(\"=\" * 84)\n",
" print(\" PROPAGATION: WIE VIEL ARBEIT SPART SICH DER SOLVER?\")\n",
" print(\" Aufgabe: n Zahlen aus 1..G, alle verschieden, Summe = 3n\")\n",
" print(\"=\" * 84)\n",
" print(f\"{'n':>3} {'G':>4} | {'Brute Force':>14} | {'CP-SAT':>10} | \"\n",
" f\"{'Ersparnis':>11} | {'Loesungen':>10}\")\n",
" print(\"-\" * 84)\n",
"\n",
" for n, grenze in [(4, 8), (5, 10), (6, 12), (7, 14)]:\n",
" kombis, treffer_bf = rohe_aufzaehlung(n, grenze)\n",
" treffer_cp, verzweigungen, dauer = mit_cp_sat(n, grenze)\n",
" assert treffer_bf == treffer_cp, \"Beide Verfahren muessen gleich viele finden!\"\n",
" ersparnis = kombis / max(verzweigungen, 1)\n",
" print(f\"{n:>3} {grenze:>4} | {kombis:>14,} | {verzweigungen:>10,} | \"\n",
" f\"{ersparnis:>10.0f}x | {treffer_cp:>10,}\")\n",
"\n",
" print(\"-\" * 84)\n",
" print(\"'Brute Force' = alle Kombinationen. 'CP-SAT' = tatsaechliche Verzweigungen.\")\n",
" print(\"Der Rest wurde durch Propagation ausgeschlossen, ohne ihn anzusehen.\")\n",
" print(\"=\" * 84)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Praxisfall: Dynamisches Vertretungssystem\n",
"\n",
"`CP_SAT_Vertretungssystem.py`\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#!/usr/bin/env python3\n",
"\n",
"# CP_SAT_Vertretungssystem.py\n",
"\"\"\"\n",
"Kapitel CP-SAT: Dynamisches Vertretungs- und Einsatzplanungssystem mit CP-SAT.\n",
"\n",
"Eigenschaften:\n",
" * Solver-Status wird ueber SolverStatus aus or_kern.py ausgewertet - dieselbe\n",
" Fallunterscheidung wie bei jedem anderen Solver im Buch\n",
" * Strafkosten-Zerlegung wird ausgewiesen (Erklaerbarkeit, Kapitel Praxisfallen)\n",
" * Fairness-Kriterium ergaenzt (Spannweite der Belastung minimieren)\n",
" * Vorabdiagnose auf offensichtliche Unloesbarkeit\n",
" * Abnahmepruefung des fertigen Plans OHNE den Solver zu fragen\n",
"\n",
"Benoetigt: ortools, pandas, pydantic (ueber or_kern)\n",
"\"\"\"\n",
"\n",
"from __future__ import annotations\n",
"\n",
"from ortools.sat.python import cp_model\n",
"import pandas as pd\n",
"\n",
"from or_kern import Loesung, SolverStatus, status_von_cpsat\n",
"\n",
"# --- 1. Datenbasis --------------------------------------------------------\n",
"SLOTS = [1, 2, 3, 4]\n",
"FAECHER = [\"Mathematik\", \"Physik\", \"Mathematik\", \"Informatik\"]\n",
"KLASSEN = [\"Klasse 8a\", \"Klasse 10b\", \"Klasse 7a\", \"Klasse 9c\"]\n",
"\n",
"PERSONAL = [\"Frau_Mueller\", \"Herr_Schmidt\", \"Frau_Albrecht\", \"Herr_Bauer\", \"Frau_Koch\"]\n",
"\n",
"QUALIFIKATION = {\n",
" \"Frau_Mueller\": {\"Mathematik\", \"Physik\"},\n",
" \"Herr_Schmidt\": {\"Informatik\", \"Mathematik\"},\n",
" \"Frau_Albrecht\": {\"Physik\", \"Informatik\"},\n",
" \"Herr_Bauer\": {\"Mathematik\"},\n",
" \"Frau_Koch\": {\"Mathematik\", \"Physik\", \"Informatik\"},\n",
"}\n",
"\n",
"VORBELASTUNG = { # bereits geleistete Stunden heute\n",
" \"Frau_Mueller\": 1, \"Herr_Schmidt\": 0, \"Frau_Albrecht\": 2,\n",
" \"Herr_Bauer\": 0, \"Frau_Koch\": 1,\n",
"}\n",
"\n",
"MAX_VERTRETUNGEN = 2\n",
"STRAFE_VORBELASTUNG = 50 # je Stunde Vorbelastung und Zuweisung\n",
"STRAFE_LOCH = 80 # je Freistundenloch\n",
"STRAFE_UNFAIRNESS = 30 # je Einheit Spannweite der Gesamtbelastung\n",
"\n",
"\n",
"def pruefe_grundsaetzliche_loesbarkeit() -> bool:\n",
" \"\"\"Vorabdiagnose: Gibt es fuer jeden Slot ueberhaupt qualifiziertes Personal?\"\"\"\n",
" ok = True\n",
" for s, fach in enumerate(FAECHER):\n",
" kandidaten = [p for p in PERSONAL if fach in QUALIFIKATION[p]]\n",
" if not kandidaten:\n",
" print(f\" UNLOESBAR: Fuer Slot {SLOTS[s]} ({fach}) gibt es niemanden.\")\n",
" ok = False\n",
" kapazitaet = len(PERSONAL) * MAX_VERTRETUNGEN\n",
" if kapazitaet < len(SLOTS):\n",
" print(f\" UNLOESBAR: Kapazitaet {kapazitaet} < {len(SLOTS)} offene Stunden.\")\n",
" ok = False\n",
" return ok\n",
"\n",
"\n",
"def pruefe_plan(plan: dict[int, str]) -> list[str]:\n",
" \"\"\"Prueft den fertigen Plan gegen die harten Regeln - ohne den Solver.\n",
"\n",
" Der Solver kann nur pruefen, was ihm gesagt wurde. Diese Funktion prueft\n",
" gegen die ANFORDERUNG und benutzt dafuer bewusst keine Modellvariable\n",
" (siehe Kapitel Praxisfallen). Leere Liste heisst bestanden.\n",
" \"\"\"\n",
" beanstandungen: list[str] = []\n",
" if sorted(plan) != list(range(len(SLOTS))):\n",
" beanstandungen.append(f\"nicht jede Stunde genau einmal besetzt: {sorted(plan)}\")\n",
" return beanstandungen\n",
" for s, person in plan.items():\n",
" if FAECHER[s] not in QUALIFIKATION[person]:\n",
" beanstandungen.append(f\"{person} ist nicht fuer {FAECHER[s]} qualifiziert\")\n",
" for person in PERSONAL:\n",
" anzahl = sum(1 for p in plan.values() if p == person)\n",
" if anzahl > MAX_VERTRETUNGEN:\n",
" beanstandungen.append(f\"{person} hat {anzahl} Vertretungen \"\n",
" f\"(hoechstens {MAX_VERTRETUNGEN})\")\n",
" return beanstandungen\n",
"\n",
"\n",
"def plane_vertretung() -> list[dict[str, str]] | None:\n",
" if not pruefe_grundsaetzliche_loesbarkeit():\n",
" return None\n",
"\n",
" modell = cp_model.CpModel()\n",
"\n",
" # Entscheidungsvariablen: x[person, slot] = 1 <=> Person uebernimmt Slot\n",
" x = {(p, s): modell.NewBoolVar(f\"zuweisung_{p}_slot{s+1}\")\n",
" for p in PERSONAL for s in range(len(SLOTS))}\n",
"\n",
" # --- HARTE NEBENBEDINGUNGEN ------------------------------------------\n",
" # H1: Jede Stunde genau einmal besetzen\n",
" for s in range(len(SLOTS)):\n",
" modell.AddExactlyOne(x[p, s] for p in PERSONAL)\n",
"\n",
" # H2: Qualifikation - unqualifizierte Zuweisung ausschliessen\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",
" # H3: Hoechstens MAX_VERTRETUNGEN Stunden je Person\n",
" for p in PERSONAL:\n",
" modell.Add(sum(x[p, s] for s in range(len(SLOTS))) <= MAX_VERTRETUNGEN)\n",
"\n",
" # (H4 \"nicht an zwei Orten gleichzeitig\" ist durch H1 bereits erfuellt:\n",
" # jeder Slot hat genau eine Person, und eine Person kann pro Slot nur\n",
" # eine Variable auf 1 setzen.)\n",
"\n",
" # --- WEICHE ZIELE ----------------------------------------------------\n",
" strafterme = []\n",
"\n",
" # W1: Vorbelastete Personen schonen\n",
" strafe_last = []\n",
" for p in PERSONAL:\n",
" zuweisungen = sum(x[p, s] for s in range(len(SLOTS)))\n",
" strafe_last.append(VORBELASTUNG[p] * STRAFE_VORBELASTUNG * zuweisungen)\n",
" strafterme.extend(strafe_last)\n",
"\n",
" # W2: Freistundenloecher vermeiden (Muster: arbeitet, frei, arbeitet)\n",
" loch_variablen = []\n",
" for p in PERSONAL:\n",
" for s in range(len(SLOTS) - 2):\n",
" loch = modell.NewBoolVar(f\"loch_{p}_{s}\")\n",
" # Reifizierung: loch == 1 <=> (x_s AND NOT x_{s+1} AND x_{s+2})\n",
" modell.AddBoolAnd([x[p, s], x[p, s + 1].Not(), x[p, s + 2]]).OnlyEnforceIf(loch)\n",
" modell.AddBoolOr([x[p, s].Not(), x[p, s + 1], x[p, s + 2].Not()]) \\\n",
" .OnlyEnforceIf(loch.Not())\n",
" loch_variablen.append(loch)\n",
" strafterme.append(loch * STRAFE_LOCH)\n",
"\n",
" # W3: Fairness - Spannweite der Gesamtbelastung minimieren\n",
" gesamtlast = {}\n",
" for p in PERSONAL:\n",
" last = modell.NewIntVar(0, len(SLOTS) + max(VORBELASTUNG.values()), f\"last_{p}\")\n",
" modell.Add(last == VORBELASTUNG[p] + sum(x[p, s] for s in range(len(SLOTS))))\n",
" gesamtlast[p] = last\n",
" max_last = modell.NewIntVar(0, 10, \"max_last\")\n",
" min_last = modell.NewIntVar(0, 10, \"min_last\")\n",
" modell.AddMaxEquality(max_last, list(gesamtlast.values()))\n",
" modell.AddMinEquality(min_last, list(gesamtlast.values()))\n",
" spannweite = modell.NewIntVar(0, 10, \"spannweite\")\n",
" modell.Add(spannweite == max_last - min_last)\n",
" strafterme.append(spannweite * STRAFE_UNFAIRNESS)\n",
"\n",
" modell.Minimize(sum(strafterme))\n",
"\n",
" # --- LOESEN -----------------------------------------------------------\n",
" loeser = cp_model.CpSolver()\n",
" loeser.parameters.max_time_in_seconds = 5.0\n",
" # Reproduzierbarkeit: ein Arbeiter, fester Startwert. Ohne das kann CP-SAT\n",
" # bei mehreren gleich guten Plaenen von Lauf zu Lauf einen anderen liefern\n",
" # (siehe JobShop_Intervalle.py im naechsten Abschnitt).\n",
" loeser.parameters.num_workers = 1\n",
" loeser.parameters.random_seed = 1\n",
" rohstatus = loeser.Solve(modell)\n",
"\n",
" # Der Rueckgabewert wird in die gemeinsame Sprache uebersetzt. Ab hier\n",
" # sieht die Auswertung genauso aus wie bei HiGHS, GLOP oder CVXPY.\n",
" status = status_von_cpsat(rohstatus)\n",
" loesung = Loesung(\n",
" status=status,\n",
" zielwert=loeser.ObjectiveValue() if status.brauchbar else None,\n",
" schranke=loeser.BestObjectiveBound() if status.brauchbar else None,\n",
" laufzeit=loeser.WallTime())\n",
"\n",
" print(\"=\" * 78)\n",
" print(\" DYNAMISCHER VERTRETUNGSPLAN (CP-SAT OPTIMIERT)\")\n",
" print(\"=\" * 78)\n",
"\n",
" if status.modellfehler:\n",
" print(f\"Kein zulaessiger Plan moeglich (Status: {status.value}).\")\n",
" print(\"Die harten Regeln widersprechen sich. Naechster Schritt: eine Regel\")\n",
" print(\"weich machen (Kapitel Praxisfallen) oder den Konflikt eingrenzen\")\n",
" print(\"(Anhang Fehlerdiagnose).\")\n",
" return None\n",
" if not status.brauchbar:\n",
" print(f\"Solver lieferte kein Ergebnis (Status: {status.value}).\")\n",
" print(\"Zeitlimit erhoehen oder das Modell vereinfachen.\")\n",
" return None\n",
"\n",
" guete = (\"beweisbar optimal\" if status is SolverStatus.OPTIMAL\n",
" else \"zulaessig, aber nicht bewiesen - Gap beachten\")\n",
" print(f\"Solver-Status: {loeser.StatusName(rohstatus)} -> {status.value} ({guete})\")\n",
" print(f\"{loesung.als_bericht()}\\n\")\n",
"\n",
" # --- Plan ausgeben ----------------------------------------------------\n",
" zuweisung = {s: next(p for p in PERSONAL if loeser.Value(x[p, s]) == 1)\n",
" for s in range(len(SLOTS))}\n",
" plan = [{\n",
" \"Stunde\": f\"Std. {SLOTS[s]}\",\n",
" \"Klasse\": KLASSEN[s],\n",
" \"Fach\": FAECHER[s],\n",
" \"Vertretung\": zuweisung[s],\n",
" \"Vorbelastung\": f\"{VORBELASTUNG[zuweisung[s]]} Std.\",\n",
" } for s in range(len(SLOTS))]\n",
" print(pd.DataFrame(plan).to_string(index=False))\n",
"\n",
" # --- Abnahmepruefung gegen die Anforderung ----------------------------\n",
" beanstandungen = pruefe_plan(zuweisung)\n",
" if beanstandungen:\n",
" print(\"\\nABNAHMEPRUEFUNG FEHLGESCHLAGEN:\")\n",
" for beanstandung in beanstandungen:\n",
" print(f\" - {beanstandung}\")\n",
" return None\n",
" print(\"\\nAbnahmepruefung: bestanden (Besetzung, Qualifikation, Hoechstzahl).\")\n",
"\n",
" # --- Erklaerbarkeit: Strafkosten aufschluesseln -----------------------\n",
" kosten_last = sum(VORBELASTUNG[p] * STRAFE_VORBELASTUNG\n",
" * sum(loeser.Value(x[p, s]) for s in range(len(SLOTS)))\n",
" for p in PERSONAL)\n",
" kosten_loecher = sum(loeser.Value(v) for v in loch_variablen) * STRAFE_LOCH\n",
" kosten_fairness = loeser.Value(spannweite) * STRAFE_UNFAIRNESS\n",
"\n",
" print(\"\\n--- Woraus bestehen die Strafkosten? ---\")\n",
" print(f\" Vorbelastung geschont: {kosten_last:5.0f} Punkte\")\n",
" print(f\" Freistundenloecher: {kosten_loecher:5.0f} Punkte \"\n",
" f\"({sum(loeser.Value(v) for v in loch_variablen)} Loch/Loecher)\")\n",
" print(f\" Fairness (Spannweite {loeser.Value(spannweite)}): {kosten_fairness:5.0f} Punkte\")\n",
" print(f\" {'Summe':<23} {loeser.ObjectiveValue():5.0f} Punkte\")\n",
"\n",
" print(\"\\n--- Auslastung nach Optimierung ---\")\n",
" for p in PERSONAL:\n",
" heute = sum(loeser.Value(x[p, s]) for s in range(len(SLOTS)))\n",
" print(f\" * {p:<15}: {VORBELASTUNG[p]} vorher + {heute} Vertretung \"\n",
" f\"= {VORBELASTUNG[p] + heute} Stunden\")\n",
"\n",
" print(f\"\\nSuchstatistik: {loeser.NumBranches()} Verzweigungen, \"\n",
" f\"{loeser.NumConflicts()} Konflikte, {loeser.WallTime():.3f} s\")\n",
" print(\"=\" * 78)\n",
" return plan\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" plane_vertretung()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Intervallvariablen: Job-Shop-Scheduling\n",
"\n",
"`JobShop_Intervalle.py`\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#!/usr/bin/env python3\n",
"\n",
"# JobShop_Intervalle.py\n",
"\"\"\"\n",
"Kapitel CP-SAT: Job-Shop-Scheduling mit Intervallvariablen.\n",
"Zeigt NewIntervalVar, AddNoOverlap und die Minimierung des Makespan.\n",
"\"\"\"\n",
"\n",
"import collections\n",
"\n",
"from ortools.sat.python import cp_model\n",
"\n",
"# (Maschine, Dauer) je Arbeitsgang, in der Reihenfolge des Auftrags\n",
"AUFTRAEGE = [\n",
" [(0, 3), (1, 2), (2, 2)], # Auftrag 0\n",
" [(0, 2), (2, 1), (1, 4)], # Auftrag 1\n",
" [(1, 4), (2, 3)], # Auftrag 2\n",
"]\n",
"MASCHINENNAMEN = [\"Fraese\", \"Dreherei\", \"Lackiererei\"]\n",
"\n",
"\n",
"def loese_jobshop():\n",
" anzahl_maschinen = 1 + max(m for auftrag in AUFTRAEGE for m, _ in auftrag)\n",
" horizont = sum(dauer for auftrag in AUFTRAEGE for _, dauer in auftrag)\n",
"\n",
" modell = cp_model.CpModel()\n",
" Gang = collections.namedtuple(\"Gang\", \"start ende intervall\")\n",
" plaene = {}\n",
" maschinen_intervalle = collections.defaultdict(list)\n",
"\n",
" # --- Intervallvariablen anlegen --------------------------------------\n",
" for a, auftrag in enumerate(AUFTRAEGE):\n",
" for g, (maschine, dauer) in enumerate(auftrag):\n",
" start = modell.NewIntVar(0, horizont, f\"start_{a}_{g}\")\n",
" ende = modell.NewIntVar(0, horizont, f\"ende_{a}_{g}\")\n",
" # Ein Intervall koppelt start + dauer == ende automatisch\n",
" intervall = modell.NewIntervalVar(start, dauer, ende, f\"intervall_{a}_{g}\")\n",
" plaene[a, g] = Gang(start, ende, intervall)\n",
" maschinen_intervalle[maschine].append(intervall)\n",
"\n",
" # --- H1: eine Maschine bearbeitet nur einen Gang gleichzeitig --------\n",
" for maschine in range(anzahl_maschinen):\n",
" modell.AddNoOverlap(maschinen_intervalle[maschine])\n",
"\n",
" # --- H2: Reihenfolge innerhalb eines Auftrags einhalten -------------\n",
" for a, auftrag in enumerate(AUFTRAEGE):\n",
" for g in range(len(auftrag) - 1):\n",
" modell.Add(plaene[a, g + 1].start >= plaene[a, g].ende)\n",
"\n",
" # --- Ziel: Makespan minimieren ---------------------------------------\n",
" makespan = modell.NewIntVar(0, horizont, \"makespan\")\n",
" modell.AddMaxEquality(makespan, [plaene[a, len(auftrag) - 1].ende\n",
" for a, auftrag in enumerate(AUFTRAEGE)])\n",
" modell.Minimize(makespan)\n",
"\n",
" loeser = cp_model.CpSolver()\n",
" loeser.parameters.max_time_in_seconds = 10.0\n",
" # Es gibt mehrere Plaene mit demselben kuerzesten Makespan. Welchen CP-SAT\n",
" # findet, haengt sonst davon ab, welcher seiner parallelen Suchstraenge\n",
" # zuerst fertig wird - dasselbe Programm liefert dann von Lauf zu Lauf\n",
" # verschiedene (gleich gute) Plaene. Fuer ein reproduzierbares Buchbeispiel\n",
" # fixieren wir beides. Im Produktivbetrieb laesst man die Standardwerte\n",
" # stehen: Mehrere Arbeiter sind dort deutlich schneller.\n",
" loeser.parameters.num_workers = 1\n",
" loeser.parameters.random_seed = 1\n",
" status = loeser.Solve(modell)\n",
"\n",
" if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):\n",
" print(f\"Kein Plan gefunden: {loeser.StatusName(status)}\")\n",
" return\n",
"\n",
" print(\"=\" * 74)\n",
" print(\" JOB-SHOP-SCHEDULING MIT INTERVALLVARIABLEN\")\n",
" print(\"=\" * 74)\n",
" print(f\"Status: {loeser.StatusName(status)} | \"\n",
" f\"Kuerzeste Gesamtdauer (Makespan): {loeser.Value(makespan)} Zeiteinheiten\\n\")\n",
"\n",
" # --- Gantt-Diagramm als Text ------------------------------------------\n",
" dauer_gesamt = loeser.Value(makespan)\n",
" print(f\"{'Maschine':<14}|\" + \"\".join(f\"{t:<3}\" for t in range(dauer_gesamt)))\n",
" print(\"-\" * (15 + 3 * dauer_gesamt))\n",
" for maschine in range(anzahl_maschinen):\n",
" zeile = [\" . \"] * dauer_gesamt\n",
" for a, auftrag in enumerate(AUFTRAEGE):\n",
" for g, (m, dauer) in enumerate(auftrag):\n",
" if m == maschine:\n",
" beginn = loeser.Value(plaene[a, g].start)\n",
" for t in range(beginn, beginn + dauer):\n",
" zeile[t] = f\" A{a}\"\n",
" print(f\"{MASCHINENNAMEN[maschine]:<14}|\" + \"\".join(zeile))\n",
"\n",
" print(\"\\n--- Detailplan ---\")\n",
" for a, auftrag in enumerate(AUFTRAEGE):\n",
" teile = []\n",
" for g, (maschine, dauer) in enumerate(auftrag):\n",
" beginn = loeser.Value(plaene[a, g].start)\n",
" teile.append(f\"{MASCHINENNAMEN[maschine]} {beginn}-{beginn + dauer}\")\n",
" print(f\" Auftrag {a}: \" + \" -> \".join(teile))\n",
"\n",
" # --- Untere Schranke zum Vergleich ------------------------------------\n",
" laengster_auftrag = max(sum(d for _, d in a) for a in AUFTRAEGE)\n",
" hoechste_maschinenlast = max(\n",
" sum(d for auftrag in AUFTRAEGE for m, d in auftrag if m == maschine)\n",
" for maschine in range(anzahl_maschinen))\n",
" schranke = max(laengster_auftrag, hoechste_maschinenlast)\n",
" print(f\"\\nUntere Schranke (laengster Auftrag / hoechste Maschinenlast): {schranke}\")\n",
" print(f\"Erreichter Makespan: {loeser.Value(makespan)}\"\n",
" f\"{' -> beweisbar bestmoeglich' if loeser.Value(makespan) == schranke else ''}\")\n",
" print(\"=\" * 74)\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" loese_jobshop()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Parallele Suche: was `num_workers` wirklich bewirkt\n",
"\n",
"`Parallele_Suche.py`\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#!/usr/bin/env python3\n",
"\n",
"# Parallele_Suche.py\n",
"\"\"\"\n",
"Kapitel CP-SAT: Was mehrere Arbeiter wirklich bringen - und was sie kosten.\n",
"\n",
"Das Buch setzt an mehreren Stellen 'num_workers = 1', damit die abgedruckte\n",
"Ausgabe reproduzierbar ist. Behauptet wird dabei zweierlei: dass parallele\n",
"Suche schneller ist, und dass sie von Lauf zu Lauf verschiedene, gleich gute\n",
"Loesungen findet. Beides wird hier gemessen statt geglaubt.\n",
"\n",
"Gerechnet wird auf einem Job-Shop wie in JobShop_Intervalle.py, nur gross\n",
"genug, dass die Suche wirklich arbeitet: 12 Auftraege auf 10 Maschinen,\n",
"jeder Auftrag besucht jede Maschine.\n",
"\n",
"Zwei Messungen:\n",
" 1. Dieselbe Aufgabe mit 1, 2, 4 und 8 Arbeitern - Laufzeit und Ergebnis.\n",
" 2. Dieselbe Konfiguration mehrfach, bei FESTEM random_seed - wie oft kommt\n",
" derselbe Plan heraus?\n",
"\n",
"Die zweite ist die wichtigere: Sie entscheidet, wie man Tests schreibt.\n",
"\n",
"Achtung, Laufzeiten sind hardwareabhaengig. Die ZAHL der verschiedenen\n",
"Plaene ist es auch - und genau das ist die Aussage.\n",
"\n",
"Benoetigt: numpy, ortools\n",
"\"\"\"\n",
"\n",
"from __future__ import annotations\n",
"\n",
"import collections\n",
"import time\n",
"\n",
"import numpy as np\n",
"from ortools.sat.python import cp_model\n",
"\n",
"AUFTRAEGE = 12\n",
"MASCHINEN = 10\n",
"INSTANZ_SEED = 20260908\n",
"SOLVER_SEED = 1 # bleibt ueber ALLE Laeufe gleich - das ist der Punkt\n",
"ARBEITERZAHLEN = (1, 2, 4, 8)\n",
"WIEDERHOLUNGEN = 4\n",
"ZEITLIMIT = 120.0\n",
"\n",
"\n",
"def baue_instanz() -> list[list[tuple[int, int]]]:\n",
" \"\"\"Klassischer Job-Shop: Jeder Auftrag besucht jede Maschine genau einmal,\n",
" in einer eigenen zufaelligen Reihenfolge.\"\"\"\n",
" rng = np.random.default_rng(INSTANZ_SEED)\n",
" return [[(int(m), int(rng.integers(2, 20))) for m in rng.permutation(MASCHINEN)]\n",
" for _ in range(AUFTRAEGE)]\n",
"\n",
"\n",
"def loese(auftraege, arbeiter: int):\n",
" \"\"\"Minimiert den Makespan. Gibt Status, Zielwert, Laufzeit und den Plan\n",
" zurueck - den Plan als Tupel aller Startzeiten, damit sich zwei Laeufe\n",
" vergleichen lassen.\"\"\"\n",
" horizont = sum(dauer for auftrag in auftraege for _, dauer in auftrag)\n",
" modell = cp_model.CpModel()\n",
" Gang = collections.namedtuple(\"Gang\", \"start ende intervall\")\n",
" plaene: dict[tuple[int, int], Gang] = {}\n",
" je_maschine = collections.defaultdict(list)\n",
"\n",
" for a, auftrag in enumerate(auftraege):\n",
" for g, (maschine, dauer) in enumerate(auftrag):\n",
" start = modell.NewIntVar(0, horizont, f\"start_{a}_{g}\")\n",
" ende = modell.NewIntVar(0, horizont, f\"ende_{a}_{g}\")\n",
" intervall = modell.NewIntervalVar(start, dauer, ende, f\"iv_{a}_{g}\")\n",
" plaene[a, g] = Gang(start, ende, intervall)\n",
" je_maschine[maschine].append(intervall)\n",
"\n",
" for maschine in range(MASCHINEN):\n",
" modell.AddNoOverlap(je_maschine[maschine])\n",
" for a, auftrag in enumerate(auftraege):\n",
" for g in range(len(auftrag) - 1):\n",
" modell.Add(plaene[a, g + 1].start >= plaene[a, g].ende)\n",
"\n",
" makespan = modell.NewIntVar(0, horizont, \"makespan\")\n",
" modell.AddMaxEquality(\n",
" makespan, [plaene[a, len(auftrag) - 1].ende\n",
" for a, auftrag in enumerate(auftraege)])\n",
" modell.Minimize(makespan)\n",
"\n",
" loeser = cp_model.CpSolver()\n",
" loeser.parameters.num_workers = arbeiter\n",
" loeser.parameters.random_seed = SOLVER_SEED\n",
" loeser.parameters.max_time_in_seconds = ZEITLIMIT\n",
"\n",
" beginn = time.perf_counter()\n",
" status = loeser.Solve(modell)\n",
" dauer = time.perf_counter() - beginn\n",
"\n",
" if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):\n",
" return loeser.StatusName(status), None, dauer, None\n",
" plan = tuple(loeser.Value(plaene[a, g].start)\n",
" for a, auftrag in enumerate(auftraege)\n",
" for g in range(len(auftrag)))\n",
" return loeser.StatusName(status), int(loeser.ObjectiveValue()), dauer, plan\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" auftraege = baue_instanz()\n",
" gaenge = sum(len(a) for a in auftraege)\n",
"\n",
" print(\"=\" * 78)\n",
" print(\" WAS MEHRERE ARBEITER WIRKLICH BRINGEN\")\n",
" print(\"=\" * 78)\n",
" print(f\"Job-Shop: {AUFTRAEGE} Auftraege, {MASCHINEN} Maschinen, \"\n",
" f\"{gaenge} Arbeitsgaenge.\")\n",
" print(f\"random_seed = {SOLVER_SEED} in JEDEM Lauf, \"\n",
" f\"{WIEDERHOLUNGEN} Wiederholungen je Arbeiterzahl.\\n\")\n",
"\n",
" # Jede Konfiguration mehrfach - ein einzelner Zeitwert sagt nichts.\n",
" ergebnisse: dict[int, list[tuple]] = {}\n",
" for arbeiter in ARBEITERZAHLEN:\n",
" ergebnisse[arbeiter] = [loese(auftraege, arbeiter)\n",
" for _ in range(WIEDERHOLUNGEN)]\n",
"\n",
" print(\"=\" * 78)\n",
" print(\" (1) Laufzeit: was die Arbeiter bringen\")\n",
" print(\"=\" * 78)\n",
" print(f\"{'Arbeiter':>9} {'Status':>9} {'Makespan':>9} \"\n",
" f\"{'schnellste':>11} {'langsamste':>11} {'Beschleunigung':>15}\")\n",
" print(\"-\" * 78)\n",
" basis_zeit = min(d for _, _, d, _ in ergebnisse[1])\n",
" for arbeiter in ARBEITERZAHLEN:\n",
" laeufe = ergebnisse[arbeiter]\n",
" zeiten = [d for _, _, d, _ in laeufe]\n",
" ziele = {o for _, o, _, _ in laeufe}\n",
" status = {s for s, _, _, _ in laeufe}\n",
" print(f\"{arbeiter:>9} {'/'.join(sorted(status)):>9} \"\n",
" f\"{'/'.join(str(z) for z in sorted(ziele)):>9} \"\n",
" f\"{min(zeiten):>10.2f}s {max(zeiten):>10.2f}s \"\n",
" f\"{basis_zeit / min(zeiten):>14.1f}x\")\n",
"\n",
" beste = min(ARBEITERZAHLEN, key=lambda w: min(d for _, _, d, _ in ergebnisse[w]))\n",
" faktor = basis_zeit / min(d for _, _, d, _ in ergebnisse[beste])\n",
" print(f\"\\n Am schnellsten sind {beste} Arbeiter: Faktor {faktor:.1f} gegenueber\")\n",
" print(f\" einem einzigen - also MEHR, als {beste} Kerne hergeben sollten.\")\n",
" print(f\" Das ist kein Messfehler. CP-SAT laesst nicht {beste}-mal dieselbe Suche\")\n",
" print(\" laufen, sondern verschiedene Strategien nebeneinander, die einander\")\n",
" print(\" ihre Schranken zurufen. Der zweite Arbeiter ist deshalb nicht der\")\n",
" print(\" 'zweite Kern', sondern ein anderes Verfahren.\")\n",
" letzte = ARBEITERZAHLEN[-1]\n",
" if beste != letzte:\n",
" zeit_letzte = min(d for _, _, d, _ in ergebnisse[letzte])\n",
" print(f\"\\n Und mehr ist nicht immer besser: {letzte} Arbeiter brauchen \"\n",
" f\"{zeit_letzte:.2f}s\")\n",
" print(f\" gegenueber {min(d for _, _, d, _ in ergebnisse[beste]):.2f}s bei \"\n",
" f\"{beste}. Ab einer gewissen Zahl kosten Abstimmung\")\n",
" print(\" und Speicherbandbreite mehr, als ein weiterer Suchstrang einbringt.\")\n",
"\n",
" print(\"\\n\" + \"=\" * 78)\n",
" print(\" (2) Reproduzierbarkeit: derselbe Seed, derselbe Plan?\")\n",
" print(\"=\" * 78)\n",
" print(f\"{'Arbeiter':>9} {'Laeufe':>7} {'verschiedene Plaene':>21} \"\n",
" f\"{'Makespan':>10}\")\n",
" print(\"-\" * 78)\n",
" for arbeiter in ARBEITERZAHLEN:\n",
" plaene = {p for _, _, _, p in ergebnisse[arbeiter] if p is not None}\n",
" ziele = {o for _, o, _, _ in ergebnisse[arbeiter]}\n",
" print(f\"{arbeiter:>9} {WIEDERHOLUNGEN:>7} {len(plaene):>21} \"\n",
" f\"{'/'.join(str(z) for z in sorted(ziele)):>10}\")\n",
"\n",
" einer = len({p for _, _, _, p in ergebnisse[1]})\n",
" # Ab welcher Arbeiterzahl bricht die Reproduzierbarkeit? Das ist die\n",
" # eigentliche Zahl - nicht, was acht Arbeiter anrichten.\n",
" ab = next((w for w in ARBEITERZAHLEN\n",
" if len({p for _, _, _, p in ergebnisse[w]}) > 1), None)\n",
" print(f\"\\n Mit einem Arbeiter: {einer} Plan aus {WIEDERHOLUNGEN} Laeufen.\")\n",
" if ab is not None:\n",
" viele = len({p for _, _, _, p in ergebnisse[ab]})\n",
" print(f\" Schon mit {ab} Arbeitern: {viele} verschiedene Plaene aus \"\n",
" f\"{WIEDERHOLUNGEN} Laeufen -\")\n",
" print(\" bei identischem random_seed und identischem Zielwert.\")\n",
" print(f\" Es braucht also keine acht Arbeiter, um die Reproduzierbarkeit\")\n",
" print(f\" zu verlieren. {ab} genuegen.\")\n",
" print(\"\\n Der Seed allein macht einen Lauf also NICHT reproduzierbar. Er legt\")\n",
" print(\" fest, wie ein einzelner Suchstrang wuerfelt - nicht, welcher von\")\n",
" print(\" mehreren zuerst fertig wird. Das entscheidet die Uhr.\")\n",
"\n",
" print(\"\\n\" + \"=\" * 78)\n",
" print(\" (3) Was daraus fuer Tests folgt\")\n",
" print(\"=\" * 78)\n",
" print(\" Ein Test der Form assert plan == erwarteter_plan besteht mal und\")\n",
" print(\" scheitert mal, ohne dass sich am Code etwas geaendert haette.\")\n",
" print(\" Zu pruefen sind stattdessen:\")\n",
" print(\" * der ZIELWERT (hier in allen Laeufen gleich),\")\n",
" print(\" * die Einhaltung aller Regeln (Abnahmepruefung, Kapitel Praxisfallen),\")\n",
" print(\" * und der Status - nicht die Gestalt der Loesung.\")\n",
" print(\"\\n Wer doch einen bestimmten Plan braucht - fuer eine abgedruckte\")\n",
" print(\" Ausgabe, fuer einen Regressionstest -, setzt num_workers = 1.\")\n",
" print(f\" Das kostet hier den Faktor {faktor:.1f} an Laufzeit und ist genau\")\n",
" print(\" deshalb eine Entscheidung fuer den Test, nicht fuer den Betrieb.\")\n",
" print(\"=\" * 78)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Die fünf Antworten von CP-SAT\n",
"\n",
"`CP_SAT_Statusfaelle.py`\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#!/usr/bin/env python3\n",
"\n",
"# CP_SAT_Statusfaelle.py\n",
"\"\"\"\n",
"Kapitel CP-SAT: Alle fuenf Antworten von CP-SAT - und was jede bedeutet.\n",
"\n",
"Der haeufigste Fehler beim Einsatz von CP-SAT im Betrieb ist eine Zeile wie\n",
"\n",
" if status == cp_model.OPTIMAL:\n",
" ...verwerte die Loesung...\n",
"\n",
"Sie wirft in drei von fuenf Faellen etwas weg, das man haette gebrauchen\n",
"koennen, und verschweigt in einem weiteren Fall einen Modellierungsfehler.\n",
"\n",
"Dieses Programm erzeugt jeden Statusfall absichtlich und zeigt die passende\n",
"Reaktion:\n",
"\n",
" OPTIMAL beweisbar bestmoeglich -> ausfuehren\n",
" FEASIBLE zulaessig, Zeit war um -> Gap pruefen, dann entscheiden\n",
" INFEASIBLE kein Plan existiert -> harte Regeln lockern\n",
" MODEL_INVALID das Modell ist fehlerhaft -> Programmierfehler beheben\n",
" UNKNOWN nichts gefunden, Zeit war um -> mehr Zeit oder Heuristik\n",
"\n",
"Benoetigt: numpy, ortools\n",
"\"\"\"\n",
"\n",
"from __future__ import annotations\n",
"\n",
"from dataclasses import dataclass\n",
"\n",
"import numpy as np\n",
"from ortools.sat.python import cp_model\n",
"\n",
"\n",
"@dataclass\n",
"class Auswertung:\n",
" \"\"\"Was nach einem CP-SAT-Lauf tatsaechlich feststeht.\"\"\"\n",
" status: str\n",
" hat_loesung: bool\n",
" beweisbar_optimal: bool\n",
" zielwert: float | None\n",
" schranke: float | None\n",
" gap: float | None\n",
" empfehlung: str\n",
"\n",
"\n",
"def werte_aus(loeser: cp_model.CpSolver, status: int,\n",
" mit_zielfunktion: bool) -> Auswertung:\n",
" \"\"\"Uebersetzt einen CP-SAT-Status in eine Handlungsempfehlung.\n",
"\n",
" Genau diese Funktion fehlt in den meisten Projekten. Sie ist kurz, aber\n",
" sie ist der Unterschied zwischen einem Prototyp und einem System, das\n",
" nachts ohne Aufsicht laeuft.\n",
" \"\"\"\n",
" name = loeser.StatusName(status)\n",
" hat_loesung = status in (cp_model.OPTIMAL, cp_model.FEASIBLE)\n",
"\n",
" zielwert = schranke = gap = None\n",
" if hat_loesung and mit_zielfunktion:\n",
" zielwert = loeser.ObjectiveValue()\n",
" schranke = loeser.BestObjectiveBound()\n",
" nenner = max(abs(zielwert), 1e-9)\n",
" gap = abs(zielwert - schranke) / nenner\n",
"\n",
" if status == cp_model.OPTIMAL:\n",
" empfehlung = \"Ausfuehren. Besser geht es nachweislich nicht.\"\n",
" elif status == cp_model.FEASIBLE:\n",
" empfehlung = (f\"Brauchbar. Der Plan ist hoechstens {gap * 100:.2f} % \"\n",
" f\"schlechter als das theoretisch Bestmoegliche.\"\n",
" if gap is not None else\n",
" \"Brauchbar, aber nicht als optimal bewiesen.\")\n",
" elif status == cp_model.INFEASIBLE:\n",
" empfehlung = (\"KEIN Plan existiert. Das ist eine Aussage ueber Ihre \"\n",
" \"harten Regeln, nicht ueber die Rechenzeit: Auch \"\n",
" \"unendlich viel Zeit wuerde nichts aendern. Regeln \"\n",
" \"lockern oder weich machen (Anhang C).\")\n",
" elif status == cp_model.MODEL_INVALID:\n",
" empfehlung = (\"Das MODELL ist fehlerhaft, nicht das Problem. \"\n",
" \"Typisch: eine Variable mit leerem Wertebereich. \"\n",
" \"model.Validate() nennt die Ursache.\")\n",
" else: # UNKNOWN\n",
" empfehlung = (\"Nichts gefunden - aber auch nicht widerlegt. Mehr Zeit \"\n",
" \"geben, das Modell vereinfachen oder mit einer \"\n",
" \"Heuristik vorbelegen (AddHint).\")\n",
"\n",
" return Auswertung(\n",
" status=name,\n",
" hat_loesung=hat_loesung,\n",
" beweisbar_optimal=status == cp_model.OPTIMAL,\n",
" zielwert=zielwert, schranke=schranke, gap=gap,\n",
" empfehlung=empfehlung,\n",
" )\n",
"\n",
"\n",
"def loese(baue, zeitlimit: float, mit_zielfunktion: bool) -> Auswertung:\n",
" modell = cp_model.CpModel()\n",
" baue(modell)\n",
" loeser = cp_model.CpSolver()\n",
" loeser.parameters.max_time_in_seconds = zeitlimit\n",
" # Fuer reproduzierbare Buchausgaben; im Betrieb weglassen.\n",
" loeser.parameters.num_workers = 1\n",
" loeser.parameters.random_seed = 1\n",
" status = loeser.Solve(modell)\n",
" return werte_aus(loeser, status, mit_zielfunktion)\n",
"\n",
"\n",
"# --- Die fuenf Faelle --------------------------------------------------------\n",
"\n",
"def dienstplan_loesbar(modell: cp_model.CpModel) -> None:\n",
" \"\"\"Vier Personen, fuenf Dienste, hoechstens zwei je Person - loesbar.\"\"\"\n",
" personen, dienste = 4, 5\n",
" x = [[modell.NewBoolVar(f\"x_{i}_{j}\") for j in range(dienste)]\n",
" for i in range(personen)]\n",
" for j in range(dienste):\n",
" modell.AddExactlyOne(x[i][j] for i in range(personen))\n",
" for i in range(personen):\n",
" modell.Add(sum(x[i]) <= 2)\n",
"\n",
"\n",
"def dienstplan_unloesbar(modell: cp_model.CpModel) -> None:\n",
" \"\"\"Dieselben fuenf Dienste, aber hoechstens EIN Dienst je Person.\n",
"\n",
" Vier Personen koennen zusammen hoechstens vier Dienste uebernehmen - fuer\n",
" fuenf Dienste reicht das nicht. Kein Solver der Welt findet hier etwas.\n",
" \"\"\"\n",
" personen, dienste = 4, 5\n",
" x = [[modell.NewBoolVar(f\"x_{i}_{j}\") for j in range(dienste)]\n",
" for i in range(personen)]\n",
" for j in range(dienste):\n",
" modell.AddExactlyOne(x[i][j] for i in range(personen))\n",
" for i in range(personen):\n",
" modell.Add(sum(x[i]) <= 1)\n",
"\n",
"\n",
"def modell_fehlerhaft(modell: cp_model.CpModel) -> None:\n",
" \"\"\"Eine Variable mit leerem Wertebereich: untere Schranke > obere.\n",
"\n",
" Im Alltag entsteht das durch einen Rechenfehler in den Grenzen, etwa\n",
" NewIntVar(mindestbesetzung, kapazitaet, ...) bei falsch sortierten Daten.\n",
" \"\"\"\n",
" kaputt = modell.NewIntVar(5, 2, \"leerer_bereich\")\n",
" modell.Add(kaputt >= 0)\n",
"\n",
"\n",
"def lastverteilung(modell: cp_model.CpModel) -> None:\n",
" \"\"\"120 Auftraege auf 11 Maschinen - gross genug, dass der Optimalitaets-\n",
" beweis laenger dauert als das Finden einer sehr guten Loesung.\"\"\"\n",
" rng = np.random.default_rng(3)\n",
" dauer = rng.integers(100, 9000, 120)\n",
" n, maschinen = len(dauer), 11\n",
" obergrenze = int(dauer.sum())\n",
"\n",
" x = [[modell.NewBoolVar(f\"x_{i}_{k}\") for k in range(maschinen)]\n",
" for i in range(n)]\n",
" for i in range(n):\n",
" modell.AddExactlyOne(x[i])\n",
" belegung = [modell.NewIntVar(0, obergrenze, f\"l_{k}\") for k in range(maschinen)]\n",
" for k in range(maschinen):\n",
" modell.Add(belegung[k] == sum(int(dauer[i]) * x[i][k] for i in range(n)))\n",
" makespan = modell.NewIntVar(0, obergrenze, \"makespan\")\n",
" modell.AddMaxEquality(makespan, belegung)\n",
" modell.Minimize(makespan)\n",
"\n",
"\n",
"def zahlpartition(modell: cp_model.CpModel) -> None:\n",
" \"\"\"46 grosse Zahlen exakt in zwei gleich schwere Haelften teilen.\n",
"\n",
" Ein beruehmt schweres Problem: Es gibt keine Zielfunktion, entweder es\n",
" geht auf oder nicht - und beides zu entscheiden dauert lange.\n",
" \"\"\"\n",
" rng = np.random.default_rng(1)\n",
" gewichte = rng.integers(10**6, 2 * 10**6, 46)\n",
" x = [modell.NewBoolVar(f\"x_{i}\") for i in range(len(gewichte))]\n",
" modell.Add(sum(int(gewichte[i]) * x[i] for i in range(len(gewichte)))\n",
" == int(gewichte.sum() // 2))\n",
"\n",
"\n",
"def zeige(titel: str, a: Auswertung) -> None:\n",
" print(f\"\\n{titel}\")\n",
" print(f\" Status {a.status}\")\n",
" print(f\" Loesung vorhanden {'ja' if a.hat_loesung else 'nein'}\")\n",
" if a.zielwert is not None:\n",
" print(f\" Zielwert / Schranke {a.zielwert:,.0f} / {a.schranke:,.0f}\"\n",
" f\" (Gap {a.gap * 100:.3f} %)\")\n",
" print(f\" -> {a.empfehlung}\")\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" print(\"=\" * 78)\n",
" print(\" DIE FUENF ANTWORTEN VON CP-SAT\")\n",
" print(\"=\" * 78)\n",
"\n",
" zeige(\"[1] Dienstplan, hoechstens 2 Dienste je Person\",\n",
" loese(dienstplan_loesbar, 10.0, mit_zielfunktion=False))\n",
"\n",
" zeige(\"[2] Lastverteilung 120 Auftraege / 11 Maschinen, 0,5 s Limit\",\n",
" loese(lastverteilung, 0.5, mit_zielfunktion=True))\n",
"\n",
" zeige(\"[3] Derselbe Fall mit 10 s Limit\",\n",
" loese(lastverteilung, 10.0, mit_zielfunktion=True))\n",
"\n",
" zeige(\"[4] Dienstplan, hoechstens 1 Dienst je Person\",\n",
" loese(dienstplan_unloesbar, 10.0, mit_zielfunktion=False))\n",
"\n",
" zeige(\"[5] Variable mit leerem Wertebereich\",\n",
" loese(modell_fehlerhaft, 10.0, mit_zielfunktion=False))\n",
"\n",
" zeige(\"[6] Zahlpartition mit 46 grossen Zahlen, 2 s Limit\",\n",
" loese(zahlpartition, 2.0, mit_zielfunktion=False))\n",
"\n",
" print(\"\\n\" + \"=\" * 78)\n",
" print(\" DIE DREI, DIE MAN NICHT VERWECHSELN DARF\")\n",
" print(\"=\" * 78)\n",
" print(\"INFEASIBLE Es gibt keine Loesung. Eine Aussage ueber Ihr MODELL.\")\n",
" print(\" Mehr Rechenzeit aendert daran nichts.\")\n",
" print(\"UNKNOWN Es wurde keine gefunden. Eine Aussage ueber die ZEIT.\")\n",
" print(\" Ob es eine gibt, ist offen.\")\n",
" print(\"MODEL_INVALID Das Modell ist gar kein gueltiges Modell. Eine\")\n",
" print(\" Aussage ueber Ihren CODE - immer ein Programmierfehler.\")\n",
" print()\n",
" print(\"Wer diese drei in einem 'else: return None' zusammenfasst, verliert\")\n",
" print(\"genau die Information, die zur Fehlersuche noetig waere.\")\n",
" print(\"=\" * 78)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"## Finde den Denkfehler\n",
"\n",
"`Strafgewichte.py`\n"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"#!/usr/bin/env python3\n",
"\n",
"# Strafgewichte.py\n",
"\"\"\"\n",
"Kapitel CP-SAT: Warum ein Strafgewicht keine Wichtigkeit ausdrueckt, sondern\n",
"einen Wechselkurs.\n",
"\n",
"Weiche Regeln kommen als Strafterm in die Zielfunktion. Die Frage, die dabei\n",
"regelmaessig falsch beantwortet wird, lautet: \"Wie gross muss die Strafe sein?\"\n",
"Die uebliche Antwort - \"je wichtiger, desto groesser\" - klingt vernuenftig und\n",
"fuehrt zuverlaessig zu Plaenen, die niemand unterschreiben will.\n",
"\n",
"Der Grund: Zwei Strafgewichte legen nicht Wichtigkeiten fest, sondern einen\n",
"UMRECHNUNGSKURS. Steht die Serienregel bei 1 und der Wunschfrei-Tag bei 10,\n",
"dann hat man dem Solver woertlich gesagt: \"Zehn Verstoesse gegen die\n",
"Serienregel sind so schlimm wie ein abgelehnter Wunsch.\" Er nimmt das\n",
"ernst - und handelt danach.\n",
"\n",
"Szenario: Schichtplan, 6 Personen, 14 Tage, 4 Besetzungen pro Tag.\n",
" HART jeder Tag genau 4 Personen; hoechstens 11 Dienste je Person\n",
" WEICH keine drei Dienste in Folge (Betriebsvereinbarung)\n",
" WEICH Wunschfrei-Tage einhalten (26 Antraege)\n",
"\n",
"Benoetigt: numpy, ortools\n",
"\"\"\"\n",
"\n",
"from __future__ import annotations\n",
"\n",
"import numpy as np\n",
"from ortools.sat.python import cp_model\n",
"\n",
"PERSONEN, TAGE, PRO_TAG = 6, 14, 4\n",
"MAX_DIENSTE = 11\n",
"STRAFE_WUNSCH = 10 # bleibt fest - variiert wird die Serienstrafe\n",
"\n",
"RNG = np.random.default_rng(2)\n",
"WUNSCHFREI = sorted({(p, t) for p in range(PERSONEN) for t in range(TAGE)\n",
" if RNG.random() < 0.35})\n",
"\n",
"\n",
"def plane(strafe_serie: int, zeitlimit: float = 25.0) -> dict:\n",
" \"\"\"Baut den Schichtplan mit dem angegebenen Gewicht fuer die Serienregel.\"\"\"\n",
" modell = cp_model.CpModel()\n",
" x = {(p, t): modell.NewBoolVar(f\"x_{p}_{t}\")\n",
" for p in range(PERSONEN) for t in range(TAGE)}\n",
"\n",
" # --- harte Regeln --------------------------------------------------\n",
" for t in range(TAGE):\n",
" modell.Add(sum(x[p, t] for p in range(PERSONEN)) == PRO_TAG)\n",
" for p in range(PERSONEN):\n",
" modell.Add(sum(x[p, t] for t in range(TAGE)) <= MAX_DIENSTE)\n",
"\n",
" # --- weiche Regel 1: keine drei Dienste in Folge --------------------\n",
" # serie[p, t] = 1 <=> Person p arbeitet an t, t+1 UND t+2.\n",
" # Die Ungleichung erzwingt das nur in eine Richtung; das genuegt, weil\n",
" # der Solver serie minimiert - er wuerde die Variable nie freiwillig\n",
" # auf 1 setzen.\n",
" strafterme = []\n",
" serien = []\n",
" for p in range(PERSONEN):\n",
" for t in range(TAGE - 2):\n",
" serie = modell.NewBoolVar(f\"serie_{p}_{t}\")\n",
" modell.Add(x[p, t] + x[p, t + 1] + x[p, t + 2] <= 2 + serie)\n",
" serien.append(serie)\n",
" strafterme.append(serie * strafe_serie)\n",
"\n",
" # --- weiche Regel 2: Wunschfrei ------------------------------------\n",
" wunschverstoesse = []\n",
" for p, t in WUNSCHFREI:\n",
" wunschverstoesse.append(x[p, t])\n",
" strafterme.append(x[p, t] * STRAFE_WUNSCH)\n",
"\n",
" modell.Minimize(sum(strafterme))\n",
"\n",
" loeser = cp_model.CpSolver()\n",
" loeser.parameters.max_time_in_seconds = zeitlimit\n",
" loeser.parameters.num_workers = 1 # reproduzierbare Buchausgabe\n",
" loeser.parameters.random_seed = 1\n",
" status = loeser.Solve(modell)\n",
" if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):\n",
" raise RuntimeError(f\"Kein Plan: {loeser.StatusName(status)}\")\n",
"\n",
" return {\n",
" \"status\": loeser.StatusName(status),\n",
" \"zielwert\": loeser.ObjectiveValue(),\n",
" \"serien\": sum(loeser.Value(v) for v in serien),\n",
" \"wunsch_verletzt\": sum(loeser.Value(v) for v in wunschverstoesse),\n",
" }\n",
"\n",
"\n",
"if __name__ == \"__main__\":\n",
" print(\"=\" * 78)\n",
" print(\" STRAFGEWICHTE SIND WECHSELKURSE, KEINE WICHTIGKEITEN\")\n",
" print(\"=\" * 78)\n",
" print(f\"{PERSONEN} Personen, {TAGE} Tage, {PRO_TAG} Besetzungen pro Tag \"\n",
" f\"= {TAGE * PRO_TAG} Dienste\")\n",
" print(f\"Das sind {TAGE * PRO_TAG / PERSONEN:.1f} Dienste je Person - der Plan ist eng.\")\n",
" print(f\"{len(WUNSCHFREI)} Wunschfrei-Antraege, Strafe je abgelehntem Wunsch: \"\n",
" f\"{STRAFE_WUNSCH}\\n\")\n",
"\n",
" print(f\"{'Strafe je 3er-Serie':>20} {'3er-Serien':>12} {'Wunsch verletzt':>17} \"\n",
" f\"{'Zielwert':>10}\")\n",
" print(\"-\" * 78)\n",
" ergebnisse = {}\n",
" for strafe in (1, 2, 5, 10, 30, 100):\n",
" e = plane(strafe)\n",
" ergebnisse[strafe] = e\n",
" print(f\"{strafe:>20} {e['serien']:>12} {e['wunsch_verletzt']:>17} \"\n",
" f\"{e['zielwert']:>10.0f}\")\n",
" print(\"-\" * 78)\n",
"\n",
" niedrig, hoch = ergebnisse[1], ergebnisse[30]\n",
" print(f\"\\nBei Strafe 1 baut der Solver {niedrig['serien']} Drei-Tage-Serien ein.\")\n",
" print(\"Die Betriebsvereinbarung steht im Modell - und wird trotzdem\")\n",
" print(\"systematisch verletzt. Das ist kein Fehler des Solvers: Bei einem\")\n",
" print(f\"Gewicht von 1 gegen {STRAFE_WUNSCH} lohnt sich jede Serie, solange sie\")\n",
" print(\"auch nur einen Zehntel-Wunsch rettet.\")\n",
" print(f\"\\nBei Strafe 30 sind es {hoch['serien']} Serien - dafuer werden\")\n",
" print(f\"{hoch['wunsch_verletzt']} statt {niedrig['wunsch_verletzt']} Wuensche abgelehnt.\")\n",
" print(\"Beide Plaene sind optimal. Sie beantworten nur verschiedene Fragen.\")\n",
" print()\n",
" print(\"Die Frage lautet also nie 'wie wichtig ist mir diese Regel?', sondern:\")\n",
" print(\" 'Wie viele abgelehnte Wuensche bin ich bereit zu akzeptieren,\")\n",
" print(\" um eine Drei-Tage-Serie zu vermeiden?'\")\n",
" print(\"Wer darauf keine Zahl nennen kann, hat das Problem noch nicht\")\n",
" print(\"verstanden - und sollte diese Tabelle dem Auftraggeber vorlegen,\")\n",
" print(\"statt ein Gewicht zu raten.\")\n",
" print()\n",
" print(\"Und die wichtigste Konsequenz: Eine Regel, die NIE gebrochen werden\")\n",
" print(\"darf, gehoert nicht in die Zielfunktion, sondern unter die harten\")\n",
" print(\"Nebenbedingungen. Alles, was einen Preis hat, wird irgendwann gekauft.\")\n",
" print(\"=\" * 78)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"name": "python",
"version": "3.11"
}
},
"nbformat": 4,
"nbformat_minor": 5
}