#!/usr/bin/env python3 # Optimierungsdienst.py """ Kapitel Testen: Das Modell als Dienst - und warum es kein normaler Endpunkt ist. Ein Optimierungsmodell hinter eine HTTP-Schnittstelle zu haengen sieht nach einer Fingeruebung aus. Es gibt aber einen Unterschied, der alles bestimmt: Eine gewoehnliche Anfrage dauert Millisekunden. Eine Optimierung dauert Sekunden bis Minuten - und manchmal laenger, als jemand warten will. Damit scheidet die naheliegende Bauform aus. Wer den Solver direkt im Request-Handler aufruft, baut einen Dienst, der bei der ersten grossen Instanz in einen Timeout des Reverse Proxy laeuft, und bei zehn gleichzeitigen Anfragen alle Arbeiter blockiert. Die tragfaehige Bauform ist deshalb ZWEISTUFIG: POST /plaene nimmt den Auftrag an, prueft ihn, gibt sofort 202 Accepted und eine Auftragsnummer zurueck GET /plaene/{id} sagt, wie weit es ist - und liefert am Ende die Loesung samt Status und Gap Drei Dinge, die dabei leicht untergehen und hier bewusst sichtbar sind: 1. Das Domaenenmodell aus or_kern.py ist zugleich das API-Schema. Eine unsinnige Anfrage wird von FastAPI mit 422 abgewiesen, bevor irgendein Solver startet - dieselbe Pruefung wie beim Excel-Import. 2. Die Antwort enthaelt STATUS und GAP, nicht nur Zahlen. Ein Aufrufer, der nur die Mengen bekommt, kann nicht unterscheiden, ob er ein bewiesenes Optimum oder einen Zeitlimit-Abbruch in der Hand haelt. 3. Jeder Auftrag hat ein ZEITLIMIT. Ohne das belegt eine einzige unguenstige Instanz einen Arbeiter auf unbestimmte Zeit. Dieses Programm laeuft ohne Server: Der Selbsttest unten benutzt den TestClient von FastAPI und spricht die Anwendung direkt an. Fuer den echten Betrieb steht am Ende, was sich aendert. Aufruf: python3 Optimierungsdienst.py # Selbsttest, kein Server noetig uvicorn Optimierungsdienst:app # echter Server auf Port 8000 Benoetigt: fastapi, httpx (fuer den Selbsttest), pydantic, ortools (ueber or_kern) """ from __future__ import annotations import time import uuid from concurrent.futures import ThreadPoolExecutor from enum import Enum from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from or_kern import (Loesung, Produktionsproblem, loese_mit_glop, pruefe_loesung) ZEITLIMIT_SEKUNDEN = 30.0 ARBEITER = 2 class Auftragsstand(str, Enum): WARTET = "wartet" LAEUFT = "laeuft" FERTIG = "fertig" GESCHEITERT = "gescheitert" class Auftragsantwort(BaseModel): """Was der Aufrufer beim Abholen bekommt. Bewusst NICHT nur die Mengen: 'stand' und die Felder aus 'loesung' (Status, Gap) sind der Unterschied zwischen einer Zahl und einer belastbaren Auskunft. """ id: str stand: Auftragsstand eingegangen: float laufzeit: float | None = None loesung: Loesung | None = None beanstandungen: list[str] = Field(default_factory=list) fehler: str | None = None app = FastAPI(title="Optimierungsdienst", summary="Produktionsplanung als Auftrag, nicht als Abfrage") # Fuer das Buchbeispiel: Auftragsbuch im Speicher, Arbeiter im selben Prozess. # Was daran im echten Betrieb nicht reicht, steht unten. AUFTRAEGE: dict[str, Auftragsantwort] = {} POOL = ThreadPoolExecutor(max_workers=ARBEITER) def rechne(auftrags_id: str, problem: Produktionsproblem) -> None: """Laeuft im Arbeiterthread - nie im Request-Handler.""" antwort = AUFTRAEGE[auftrags_id] antwort.stand = Auftragsstand.LAEUFT start = time.perf_counter() try: loesung = loese_mit_glop(problem) antwort.loesung = loesung # Dieselbe Abnahmepruefung wie ueberall sonst. Ein Dienst, der sie # weglaesst, liefert Fehler schneller aus als ein Mensch sie faende. antwort.beanstandungen = pruefe_loesung(problem, loesung) antwort.stand = (Auftragsstand.FERTIG if loesung.status.brauchbar and not antwort.beanstandungen else Auftragsstand.GESCHEITERT) if not loesung.status.brauchbar: antwort.fehler = f"Solverstatus: {loesung.status.value}" elif antwort.beanstandungen: antwort.fehler = "Abnahmepruefung fehlgeschlagen" except Exception as fehler: # noqa: BLE001 antwort.stand = Auftragsstand.GESCHEITERT antwort.fehler = f"{type(fehler).__name__}: {fehler}" finally: antwort.laufzeit = time.perf_counter() - start @app.post("/plaene", status_code=202, response_model=Auftragsantwort) def auftrag_annehmen(problem: Produktionsproblem) -> Auftragsantwort: """Nimmt an, prueft, gibt sofort zurueck. Der Typ 'Produktionsproblem' im Parameter ist der ganze Trick: FastAPI validiert die Anfrage damit gegen das Domaenenmodell und antwortet bei Unsinn mit 422, ohne dass hier eine Zeile Pruefcode steht. """ auftrags_id = str(uuid.uuid4()) AUFTRAEGE[auftrags_id] = Auftragsantwort( id=auftrags_id, stand=Auftragsstand.WARTET, eingegangen=time.time()) POOL.submit(rechne, auftrags_id, problem) return AUFTRAEGE[auftrags_id] @app.get("/plaene/{auftrags_id}", response_model=Auftragsantwort) def auftrag_abholen(auftrags_id: str) -> Auftragsantwort: if auftrags_id not in AUFTRAEGE: raise HTTPException(status_code=404, detail="Unbekannter Auftrag") return AUFTRAEGE[auftrags_id] @app.get("/gesundheit") def gesundheit() -> dict[str, object]: """Was ein Ueberwachungssystem abfragt - siehe Betriebsueberwachung.py.""" offen = sum(1 for a in AUFTRAEGE.values() if a.stand in (Auftragsstand.WARTET, Auftragsstand.LAEUFT)) return {"zustand": "bereit", "auftraege_gesamt": len(AUFTRAEGE), "offen": offen, "arbeiter": ARBEITER} # --- Selbsttest ohne Server ------------------------------------------------- SCHREINEREI = { "produkte": [ {"name": "Tisch", "deckungsbeitrag": 240.0, "verbrauch": {"Montagestunden": 3.0, "Plattenmaterial": 6.0}}, {"name": "Stuhl", "deckungsbeitrag": 60.0, "verbrauch": {"Montagestunden": 1.0, "Plattenmaterial": 1.0}}, ], "kapazitaeten": {"Montagestunden": 150.0, "Plattenmaterial": 240.0}, } DOCKERFILE = """\ FROM python:3.12-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY or_kern.py Optimierungsdienst.py . # Ein Arbeiter je CPU-Kern - der Solver rechnet selbst schon parallel, # mehr Prozesse machen ihn nicht schneller, sondern langsamer. CMD ["uvicorn", "Optimierungsdienst:app", "--host", "0.0.0.0", "--port", "8000"] """ def warte_auf_ergebnis(klient, auftrags_id: str, grenze: float = 30.0) -> dict: ende = time.time() + grenze while time.time() < ende: antwort = klient.get(f"/plaene/{auftrags_id}").json() if antwort["stand"] in ("fertig", "gescheitert"): return antwort time.sleep(0.02) raise TimeoutError("Auftrag wurde nicht fertig") if __name__ == "__main__": from fastapi.testclient import TestClient print("=" * 80) print(" DAS MODELL ALS DIENST") print("=" * 80) with TestClient(app) as klient: # 1. Gueltiger Auftrag print("\n1. Auftrag einreichen") angenommen = klient.post("/plaene", json=SCHREINEREI) auftrags_id = angenommen.json()["id"] print(f" POST /plaene -> {angenommen.status_code} " f"{'Accepted' if angenommen.status_code == 202 else ''}") print(f" Auftragsnummer -> UUID, {len(auftrags_id)} Zeichen") # Der entscheidende Punkt, und er ist pruefbar: Die Antwort ist da, # BEVOR es eine Loesung gibt. Genau dafuer ist 202 gedacht. print(f" Loesung schon dabei? -> " f"{'ja' if angenommen.json()['loesung'] else 'nein - genau so soll es sein'}") fertig = warte_auf_ergebnis(klient, auftrags_id) loesung = fertig["loesung"] print("\n2. Ergebnis abholen") print(f" Stand -> {fertig['stand']}") print(f" Solverstatus -> {loesung['status']}") print(f" Zielwert -> {loesung['zielwert']:,.2f} EUR") print(f" Mengen -> " + ", ".join( f"{name} {wert:.0f}" for name, wert in loesung["werte"].items())) print(f" Schattenpreise -> " + ", ".join( f"{name} {wert:.0f}" for name, wert in loesung["schattenpreise"].items())) print(f" Abnahmepruefung -> " f"{'bestanden' if not fertig['beanstandungen'] else fertig['beanstandungen']}") # 2. Unsinnige Anfrage - muss abgewiesen werden, BEVOR gerechnet wird print("\n3. Unsinnige Anfragen (der Solver startet gar nicht erst)") for beschreibung, aenderung in [ ("Kapazitaet 0", {"kapazitaeten": {"Montagestunden": 0.0}}), ("keine Produkte", {"produkte": []}), ("Ressource ohne Kapazitaet", {"produkte": [{"name": "Regal", "deckungsbeitrag": 130.0, "verbrauch": {"Lackieren": 2.0}}]})]: anfrage = {**SCHREINEREI, **aenderung} antwort = klient.post("/plaene", json=anfrage) print(f" {beschreibung:<24} -> {antwort.status_code} " f"{'Unprocessable Content' if antwort.status_code == 422 else ''}") print("\n4. Unbekannter Auftrag") print(f" GET /plaene/gibtsnicht -> " f"{klient.get('/plaene/gibtsnicht').status_code} Not Found") print("\n5. Gesundheitsabfrage") print(f" GET /gesundheit -> {klient.get('/gesundheit').json()}") # --- Warum ein Thread je Auftrag genuegt ----------------------------- print("\n" + "-" * 80) print("6. Warum Threads hier reichen (und wann nicht)\n") from concurrent.futures import ThreadPoolExecutor as Pool # Wichtig: eine Rechnung mit ECHTER Last. Ein Modell, das in zehn # Millisekunden fertig ist, misst nur den Aufwand fuers Threadstarten - # der erste Entwurf dieser Messung ist genau daran gescheitert und zeigte # eine Verlangsamung, wo in Wirklichkeit eine Beschleunigung steht. def eine_rechnung() -> None: import numpy as np from ortools.linear_solver import pywraplp rng = np.random.default_rng(3) n = 260 solver = pywraplp.Solver.CreateSolver("SCIP") x = [solver.IntVar(0, 1, f"x{i}") for i in range(n)] gewicht = rng.integers(10, 60, n) wert = rng.integers(10, 60, n) solver.Add(sum(int(gewicht[i]) * x[i] for i in range(n)) <= int(gewicht.sum() * 0.5)) for _ in range(30): auswahl = rng.choice(n, 40, replace=False) solver.Add(sum(x[int(i)] for i in auswahl) <= 12) solver.Maximize(sum(int(wert[i]) * x[i] for i in range(n))) solver.Solve() t0 = time.perf_counter() for _ in range(4): eine_rechnung() seriell = time.perf_counter() - t0 t0 = time.perf_counter() with Pool(max_workers=4) as p: list(p.map(lambda _: eine_rechnung(), range(4))) parallel = time.perf_counter() - t0 print(f" vier Rechnungen nacheinander -> {seriell:6.2f} s") print(f" vier Rechnungen in Threads -> {parallel:6.2f} s " f"(Faktor {seriell / parallel:.1f})") print() if parallel < seriell * 0.75: print(" Die Laeufe ueberlappen sich. ortools rechnet in C++ und gibt") print(" den GIL waehrend Solve() frei - deshalb genuegt hier ein") print(" Threadpool, es braucht keine eigenen Prozesse.") else: print(" Keine Ueberlappung - hier waeren Prozesse noetig.") print() print(" ACHTUNG, das gilt nicht allgemein: Eine in reinem Python") print(" geschriebene Heuristik (Kapitel Metaheuristiken) haelt den GIL") print(" die ganze Zeit. Fuer sie braucht derselbe Dienst einen") print(" ProcessPoolExecutor statt eines Threadpools.") print("\n" + "=" * 80) print(" WAS SICH IM ECHTEN BETRIEB AENDERT") print("=" * 80) print("Dieses Beispiel haelt das Auftragsbuch im Speicher und rechnet in") print("Threads desselben Prozesses. Das reicht zum Zeigen und fuer einen") print("einzelnen Rechner - nicht darueber hinaus:") print() print(" * NEUSTART LOESCHT ALLES. Auftragsbuch in eine Datenbank oder eine") print(" Warteschlange (Redis, RabbitMQ), nicht in ein dict.") print(" * ZWEI INSTANZEN KENNEN EINANDER NICHT. Sobald der Dienst mehr als") print(" einmal laeuft, muss die Warteschlange ausserhalb liegen -") print(" typischerweise Celery mit Redis als Vermittler.") print(" * DIE WAHL THREADS/PROZESSE HAENGT AM SOLVER - siehe die Messung") print(" unter Punkt 6. Sie gehoert gemessen, nicht angenommen.") print(" * OHNE ZEITLIMIT KEIN DIENST. Jede Instanz bekommt eines, und der") print(" Aufrufer erfaehrt im Status, ob es gegriffen hat.") print() print("Das zugehoerige Dockerfile ist kurz genug, um es ganz zu zeigen:") print() for zeile in DOCKERFILE.splitlines(): print(f" {zeile}") print() print("(Es wird hier nicht gebaut - das Buch setzt keine laufende") print(" Docker-Installation voraus.)") print("=" * 80)