#!/usr/bin/env python3 # KI-Liga: dein Rechner im Battle. Ein Knopf, kein Setup außer Ollama. # # python3 bench.py # passenden Test wählen, laufen lassen, hochladen # python3 bench.py --alle # alle Stufen testen, die in den Speicher passen # python3 bench.py --stufe standard # python3 bench.py --name "Jaimy" --geraet "Alienware Desktop" # sonst wird gefragt # python3 bench.py --kein-upload # nur messen, nichts senden # python3 bench.py --watt-url http://192.168.178.105/report # eigene Messdose (myStrom/Tapo-Strommesser) # # Strom wird automatisch erkannt: Mac liest macOS (SystemPowerIn), NVIDIA liest nvidia-smi (nur GPU). # Ohne beides läuft der Test trotzdem, dann zählt nur das Tempo (tok/s), nicht die Effizienz. # Nur Python-Standardbibliothek. Voraussetzung: Ollama läuft (https://ollama.com). import argparse import json import os import platform import re import secrets import socket import subprocess import sys import threading import time import urllib.error import urllib.parse import urllib.request OLLAMA = os.environ.get("OLLAMA", "http://localhost:11434") SERVER = os.environ.get("KILIGA_SERVER", "https://jaimy.net/ki-liga") TOKEN = "kiliga-2026" # nur gegen Zufalls-Spam, kein Geheimnis TOOL_VERSION = 2 # Stufen: Ollama-Tag, Mindest-Grafikspeicher (NVIDIA, GB), Mindest-Gesamtspeicher (Mac, GB). STUFEN = [ ("leicht", "qwen2.5:7b", 6, 10, "Leicht (7B)"), ("standard", "qwen2.5:14b", 11, 16, "Standard (14B)"), ("gross", "qwen3.8:27b", 22, 30, "Gross (27B)"), ] P = "Schreibe einen ausführlichen Statusbericht für ein IT-Grossprojekt mit Risiken, Maßnahmen und nächsten Schritten." def sh(*a): try: return subprocess.run(a, capture_output=True, text=True, timeout=20).stdout except Exception: return "" # ---------- Strommessung ---------- def mac_watt(): """Leistung am Netzteil laut macOS, kein sudo. Liefert None auf anderen Systemen.""" out = sh("ioreg", "-rw0", "-c", "AppleSmartBattery") tel = re.search(r'"PowerTelemetryData" = \{([^}]*)\}', out) if tel: m = re.search(r'"SystemPowerIn"=(\d+)', tel[1]) if m: return int(m[1]) / 1000 return None def nvidia_watt(): out = sh("nvidia-smi", "--query-gpu=power.draw", "--format=csv,noheader,nounits") try: return float(out.strip().splitlines()[0]) except Exception: return None def dose_watt(url): try: raw = urllib.request.urlopen(url, timeout=3).read().decode().strip() try: d = json.loads(raw) for k in ("watt", "power", "Ws", "W", "power_mw"): if k in d: return d[k] / 1000 if k == "power_mw" else float(d[k]) except Exception: return float(re.search(r"[-\d.]+", raw)[0]) except Exception: return None return None def strom_quelle(args): """Wählt die Stromquelle: eigene Dose > Mac > NVIDIA > keine.""" if args.watt_url: return ("dose", lambda: dose_watt(args.watt_url), "Messdose") if mac_watt() is not None: return ("mac", mac_watt, "Mac (macOS)") if nvidia_watt() is not None: return ("nvidia", nvidia_watt, "nur GPU (nvidia-smi)") return (None, lambda: None, None) class Meter: """Liest die Leistung jede Sekunde im Hintergrund.""" def __init__(self, read): self.read, self.vals, self.stop = read, [], threading.Event() threading.Thread(target=self._run, daemon=True).start() def _run(self): while not self.stop.is_set(): w = self.read() if w is not None: self.vals.append(w) time.sleep(1) def end(self): self.stop.set() v = self.vals return {"schnitt_w": round(sum(v) / len(v), 1) if v else None, "max_w": round(max(v), 1) if v else None, "proben": len(v)} # ---------- Ollama ---------- def api(path, body=None, timeout=900): req = urllib.request.Request(OLLAMA + path, json.dumps(body).encode() if body is not None else None) return json.load(urllib.request.urlopen(req, timeout=timeout)) def gen(model, n, salt): d = api("/api/generate", {"model": model, "stream": False, "prompt": f"[Lauf {salt}] " + P, "options": {"num_predict": n, "num_ctx": 4096}}) return {"out": d["eval_count"], "eval_s": d["eval_duration"] / 1e9} def lasttest(model, k, dur, read): end, stats, lock, cnt = time.time() + dur, [], threading.Lock(), [0] def worker(): while time.time() < end: with lock: cnt[0] += 1 salt = cnt[0] d = gen(model, 400, salt) with lock: stats.append(d) meter = Meter(read) t0 = time.time() ts = [threading.Thread(target=worker) for _ in range(k)] [t.start() for t in ts] [t.join() for t in ts] el = time.time() - t0 out = sum(d["out"] for d in stats) return {"anfragen": len(stats), "tok_s_je_anfrage": round(sum(d["out"] / d["eval_s"] for d in stats) / len(stats), 1) if stats else 0, "tok_s_gesamt": round(out / el, 1), **meter.end()} # ---------- Rechner erkennen ---------- def hardware(): sysname = platform.system() gpu_name = sh("nvidia-smi", "--query-gpu=name", "--format=csv,noheader").strip().splitlines() vram = sh("nvidia-smi", "--query-gpu=memory.total", "--format=csv,noheader,nounits").strip().splitlines() vram_gb = round(int(vram[0]) / 1024) if vram and vram[0].strip().isdigit() else None if sysname == "Darwin": chip = sh("sysctl", "-n", "machdep.cpu.brand_string").strip() ram = int(sh("sysctl", "-n", "hw.memsize") or 0) // 2**30 g = re.search(r"Total Number of Cores: (\d+)", sh("system_profiler", "SPDisplaysDataType")) return {"typ": "mac", "chip": chip, "anzeige": chip, "ram_gb": ram, "vram_gb": ram, "gpu_kerne": int(g[1]) if g else None, "os": "macOS " + platform.mac_ver()[0]} if gpu_name and gpu_name[0]: name = re.sub(r"^NVIDIA\s+(GeForce\s+)?", "", gpu_name[0]).strip() return {"typ": "nvidia", "chip": name, "anzeige": name + (f" {vram_gb} GB" if vram_gb else ""), "ram_gb": None, "vram_gb": vram_gb, "gpu_kerne": None, "os": platform.system() + " " + platform.release()} return {"typ": "pc", "chip": platform.processor() or "Unbekannt", "anzeige": (platform.processor() or "CPU") + " (ohne NVIDIA)", "ram_gb": None, "vram_gb": None, "gpu_kerne": None, "os": platform.system() + " " + platform.release()} def passende_stufen(hw): speicher = hw["vram_gb"] or hw["ram_gb"] or 0 out = [] for key, tag, min_vram, min_ram, label in STUFEN: grenze = min_ram if hw["typ"] == "mac" else min_vram if speicher == 0 or speicher >= grenze: out.append((key, tag, label)) return out # ---------- Upload ---------- def gemerkt(): try: return json.load(open(os.path.expanduser("~/.ki-liga.json"))) except Exception: return {} def merken(**werte): d = gemerkt() d.update({k: v for k, v in werte.items() if v}) try: json.dump(d, open(os.path.expanduser("~/.ki-liga.json"), "w"), ensure_ascii=False) except Exception: pass def frage(text, vorgabe): a = input(f"{text}" + (f" [{vorgabe}]" if vorgabe else "") + ": ").strip() return a or vorgabe def hochladen(server, payload): req = urllib.request.Request(server + "/submit.php", json.dumps(payload).encode(), {"Content-Type": "application/json"}) try: return json.load(urllib.request.urlopen(req, timeout=30)) except urllib.error.HTTPError as e: try: meldung = json.load(e).get("fehler") except Exception: meldung = None raise RuntimeError(meldung or f"HTTP {e.code}") from None # ---------- Ablauf ---------- def ein_lauf(hw, key, tag, label, args, quelle, read): print(f"\n== {label} · {tag} ==") have = {m["name"] for m in api("/api/tags").get("models", [])} if tag not in have and tag + ":latest" not in have: print(f" Lade {tag} herunter (einmalig, kann dauern) ...") try: api("/api/pull", {"name": tag, "stream": False}, timeout=7200) except Exception as e: print(f" Modell {tag} nicht ladbar ({e}). Stufe übersprungen.") return None gen(tag, 5, 0) # laden, nicht mitmessen print(f" Leerlauf {args.leerlauf}s (bitte Finger weg) ...") api("/api/generate", {"model": tag, "keep_alive": 0}) m = Meter(read) time.sleep(args.leerlauf) idle = m.end() gen(tag, 5, 0) # wieder laden print(f" Chat (1 Anfrage, {args.dauer}s) ...") single = lasttest(tag, 1, args.dauer, read) print(f" Stapel (4 gleichzeitig, {args.dauer}s) ...") par = lasttest(tag, 4, args.dauer, read) watt = single["schnitt_w"] token_wh = round(single["tok_s_je_anfrage"] / watt * 3600) if watt else None print(f" -> {single['tok_s_je_anfrage']} tok/s" + (f", {watt} W, {token_wh} Token/Wh" if watt else " (ohne Strommessung)")) return {"stufe": key, "modell": tag, "single_tok_s": single["tok_s_je_anfrage"], "par_tok_s": par["tok_s_gesamt"], "watt_last": watt, "watt_idle": idle["schnitt_w"], "watt_quelle": quelle, "token_wh": token_wh} def main(): ap = argparse.ArgumentParser(description="KI-Liga Benchmark") ap.add_argument("--name", help="Spitzname für die Rangliste") ap.add_argument("--geraet", help="Name des Rechners, z.B. 'Gaming-PC' oder 'MacBook Pro'") ap.add_argument("--stufe", choices=[s[0] for s in STUFEN], help="nur diese Stufe") ap.add_argument("--alle", action="store_true", help="alle passenden Stufen testen") ap.add_argument("--dauer", type=int, default=45, help="Sekunden je Lasttest (Standard 45)") ap.add_argument("--leerlauf", type=int, default=30, help="Sekunden Leerlauf-Messung (Standard 30)") ap.add_argument("--watt-url", help="URL einer eigenen Messdose (liefert Watt)") ap.add_argument("--server", default=SERVER, help="Liga-Server") ap.add_argument("--kein-upload", action="store_true", help="nur messen, nichts senden") ap.add_argument("--schluessel", help="Besitzer-Schlüssel von deinem ersten Rechner (für weitere Rechner)") a = ap.parse_args() try: api("/api/version") except Exception: raise SystemExit(f"Ollama antwortet nicht unter {OLLAMA}. Bitte Ollama starten (https://ollama.com).") hw = hardware() quelle, read, quelle_label = strom_quelle(a) print(f"Rechner: {hw['anzeige']} · {hw['os']}") print(f"Strom: {quelle_label or 'keine Messung (nur Tempo)'}") if a.stufe: stufen = [(s[0], s[1], s[4]) for s in STUFEN if s[0] == a.stufe] else: stufen = passende_stufen(hw) if not a.alle: stufen = stufen[-1:] # größte passende Stufe if not stufen: raise SystemExit("Keine passende Modellstufe für den Speicher gefunden.") print("Teste: " + ", ".join(s[2] for s in stufen)) ergebnisse = [r for s in stufen if (r := ein_lauf(hw, s[0], s[1], s[2], a, quelle, read))] if not ergebnisse: raise SystemExit("Kein Lauf erfolgreich.") paket = {"token": TOKEN, "tool_version": TOOL_VERSION, "rechner": {k: hw[k] for k in ("typ", "chip", "anzeige", "ram_gb", "vram_gb", "gpu_kerne", "os")}, "zeit": time.strftime("%Y-%m-%d %H:%M"), "laeufe": ergebnisse} lokal = os.path.join(os.path.dirname(os.path.abspath(__file__)), f"liga-ergebnis-{time.strftime('%Y%m%d-%H%M')}.json") json.dump(paket, open(lokal, "w", encoding="utf-8"), ensure_ascii=False, indent=1) if a.kein_upload: print(f"\nFertig (ohne Upload). Datei: {lokal}") return alt = gemerkt() print() name = a.name or frage("Dein Spitzname für die Rangliste", alt.get("name")) geraet = a.geraet or frage("Name dieses Rechners", alt.get("geraet") or hw["anzeige"]) if not name: print(f"Kein Name, kein Upload. Datei: {lokal}") return # Besitzer-Schlüssel: schützt den Spitznamen, bleibt auf diesem Rechner, der Server kennt nur seinen Hash. schluessel = a.schluessel or os.environ.get("KILIGA_SCHLUESSEL") or alt.get("schluessel") or secrets.token_urlsafe(24) merken(name=name, geraet=geraet, schluessel=schluessel) paket["spitzname"], paket["geraet"], paket["schluessel"] = name, geraet, schluessel try: antwort = hochladen(a.server, paket) except Exception as e: print(f"\nUpload fehlgeschlagen ({e}). Deine Datei liegt hier: {lokal}") return print("\n" + "=" * 42) for r in antwort.get("raenge", []): p = f"Platz {r['potenz_rang']}/{r['anzahl']} (Tempo)" e = f", Platz {r['effizienz_rang']}/{r['effizienz_anzahl']} (Effizienz)" if r.get("effizienz_rang") else "" print(f"{r['modell']}: {p}{e}") print(f"Rangliste: {a.server}/#{name}") print(f"Deine Spielerakte: {a.server}/#spieler={urllib.parse.quote(name)}") print("=" * 42) if antwort.get("besitz") == "neu": print(f"\nDein Spitzname „{name}“ ist jetzt geschützt. Der Schlüssel liegt in ~/.ki-liga.json.") print("Auf weiteren Rechnern unter demselben Namen:") print(f" python3 bench.py --schluessel {schluessel}") print("Den Schlüssel nicht weitergeben, er ist dein Ausweis in der Liga.") if __name__ == "__main__": socket.setdefaulttimeout(900) main()