diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..27050ae --- /dev/null +++ b/.dockerignore @@ -0,0 +1,12 @@ +.venv/ +__pycache__/ +*.pyc +.git/ +.gitignore +README.md +*.md +data/maps/*.scx +data/maps/*.scm +data/maps/*.wav +# Die Beispiel-Basis-Karte wird zur Laufzeit ueber das Volume bereitgestellt, +# nicht ins Image gebacken (Volume ueberlagert /data/maps ohnehin). diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2e5f2ea --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +# Python +.venv/ +__pycache__/ +*.pyc +*.egg-info/ +.pytest_cache/ + +# Generierte Missionen / WAVs im Karten-Verzeichnis (Basis-Karte wird behalten, +# siehe Negation unten). +data/maps/* +!data/maps/base-map.scx +!data/maps/.gitkeep diff --git a/Caddyfile b/Caddyfile new file mode 100644 index 0000000..e2cd3ad --- /dev/null +++ b/Caddyfile @@ -0,0 +1,16 @@ +# Caddy-Reverse-Proxy fuer den StarCraft-MCP-Server. +# +# Fuege diesen Block in deine bestehende Caddyfile ein (oder importiere die Datei). +# Caddy holt automatisch ein TLS-Zertifikat fuer die Subdomain. +# +# Voraussetzung: Caddy und der sc-mcp-Container haengen im selben Docker-Netzwerk +# (siehe docker-compose.yml -> networks). Dann ist der Service unter "sc-mcp:8000" +# erreichbar. + +sc-mcp.pixel-by-design.de { + reverse_proxy sc-mcp:8000 +} + +# Hinweis: Der MCP-Endpunkt (Streamable HTTP) liegt unter dem Pfad /mcp, also +# https://sc-mcp.pixel-by-design.de/mcp -- genau diese URL traegst du in Claude +# als Custom Connector ein. diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..aa8a661 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,28 @@ +FROM python:3.12-slim + +# StormLib (von RichChk mitgeliefert) braucht keine extra System-Libs auf glibc-Basis. +# 'slim' (Debian) ist glibc -> kompatibel mit der mitgelieferten StormLib .so. + +ENV PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 \ + SC_TRANSPORT=http \ + SC_HOST=0.0.0.0 \ + SC_PORT=8000 \ + SC_MAPS_DIR=/data/maps + +WORKDIR /app + +COPY requirements.txt /app/requirements.txt +RUN pip install --no-cache-dir -r requirements.txt + +COPY starcraft_mcp /app/starcraft_mcp +COPY selftest.py /app/selftest.py + +# Karten-Volume (Basis-Karten, WAVs, fertige Missionen). +RUN mkdir -p /data/maps +VOLUME ["/data/maps"] + +EXPOSE 8000 + +# Streamable-HTTP-Endpunkt unter http://0.0.0.0:8000/mcp +CMD ["python", "-m", "starcraft_mcp.server"] diff --git a/README.md b/README.md index 915782c..9f79f1e 100644 --- a/README.md +++ b/README.md @@ -1,155 +1,225 @@ # Star-Edit — StarCraft-Kampagnen-MCP-Server ("Missions-Baumeister") -> **Status:** Planung / Spezifikation. Dieses Dokument hält den kompletten Auftrag -> fest, bevor Code gebaut wird. Es dient als verbindliche Vorlage für die Umsetzung. +Ein **MCP-Server**, mit dem Claude StarCraft-Brood-War-Karten (`.scm`/`.scx`) liest und +schreibt. Du lieferst die Kreativ-Seite (Story, Dialoge, Voiceover, Sounds, Ablauf), +Claude baut daraus echte, **spielbare Missionsdateien** — komplette Logik, Texte und +Sound-Einbettung. -## Ziel in einem Satz +> **Das Gelände wird NICHT generiert.** Jede Mission startet von einer vorhandenen +> Basis-Karte als Leinwand; der Server bearbeitet nur deren Daten-/Logik-Sektionen +> (Trigger, Locations, Player-Setup, Sounds). -Ein **MCP-Server**, der Claude in die Lage versetzt, aus der Kreativ-Seite einer -StarCraft-Brood-War-Kampagne (Story, Dialoge, Voiceover, Sounds, Missionsablauf) -echte, spielbare `.scm`/`.scx`-Missionsdateien zu bauen — also komplette -Missions-Logik, Texte und Sound-Einbettung. +Die ganze Karten-Manipulation läuft über die Python-Bibliothek +[RichChk](https://github.com/sethmachine/richchk). -**Wichtig:** Das **Gelände wird nicht generiert**. Jede Mission startet von einer -vorhandenen Basis-Karte als Leinwand; der Server bearbeitet nur die Daten-/Logik- -Sektionen darin. +--- -## Rollenverteilung +## In 3 Schritten loslegen -- **Ich (Auftraggeber):** liefere die Kreativ-Seite — Story, Dialoge, Voiceover, - Sounds, Missionsablauf. Null Programmierkenntnisse. Will am Ende **einen Befehl** - ausführen und eine **URL** in Claude eintragen. -- **Claude + MCP-Server:** setzt die Kreativ-Vorlage in echte Missionsdateien um. +### 1. Server starten (ein Befehl) -## Technische Basis +Auf dem VPS, im Projektverzeichnis: -- **Karten-Manipulation:** Python-Bibliothek **RichChk** (PyPI: `richchk`). - - `pip install richchk` funktioniert; bringt StormLib mit, läuft auf Linux - (getestet — `StarCraftMpqIoHelper.create_mpq_io()` startet ohne Fehler). -- **Bestätigte Einstiegspunkte** (Signaturen vor dem Coden per Introspektion final - prüfen, da sie je nach Version leicht abweichen): +```bash +./run.sh +``` - Lesen/Schreiben von Karten (MPQ-Container): - ```python - from richchk.io.mpq.starcraft_mpq_io_helper import StarCraftMpqIoHelper - mpq_io = StarCraftMpqIoHelper.create_mpq_io() - rich_chk = mpq_io.read_chk_from_mpq("/pfad/map.scx") # -> RichChk-Modell - mpq_io.save_chk_to_mpq(rich_chk, "/pfad/basis.scx", "/pfad/out.scx") - ``` +Das baut das Docker-Image (beim ersten Mal) und startet den Container `sc-mcp`. +Er lauscht intern auf Port **8000** mit Streamable-HTTP unter dem Pfad **`/mcp`**. - Bearbeiten: - ```python - from richchk.editor.richchk.rich_chk_editor import RichChkEditor - new_chk = RichChkEditor().replace_chk_section(neue_sektion, rich_chk) - rich_chk.get_sections_by_name(...) # vorhandene Sektionen abfragen - rich_chk.chk_sections # alle Sektionen - ``` +Weitere Befehle: -- **Fertige Editoren** unter `richchk.editor.richchk.*`: - **trig** (Trigger), **mrgn** (Locations), **unis/unix** (Einheiten-Werte), - **forc** (Forces/Fraktionen), **ownr** (Owner), **side** (Rassen), - **swnm** (Switches), **uprp** (Unit Properties), **wav** (Sounds). -- **Rich-Modelle** (Trigger, Conditions, Actions, Locations, …) unter - `richchk.model.richchk.*`. +```bash +./run.sh logs # Live-Logs ansehen +./run.sh selftest # den eingebauten Selbsttest laufen lassen +./run.sh stop # Container stoppen +``` -**Herzstück = Trigger.** Fast die gesamte Kampagnen-Logik wird über Trigger -abgebildet: Gegnerwellen, Verstärkungen, Einheiten spawnen, Story-Text einblenden, -Funksprüche mit Portrait, WAV-Sounds abspielen, Missionsziele, Sieg/Niederlage, -Timer, AI-Skripte starten. Das Trigger-Tool muss ausdrucksstark und gut -dokumentiert sein (unterstützte Condition-/Action-Typen in der Tool-Beschreibung). +### 2. Subdomain via Caddy (TLS) -### Erster Arbeitsschritt (vor dem Tool-Coding) +Caddy terminiert TLS und leitet die Subdomain an den Container weiter. Trage den Block +aus der [`Caddyfile`](./Caddyfile) in deine bestehende Caddy-Konfiguration ein: -1. `richchk` installieren. -2. README/Quellen des GitHub-Repos "richchk" lesen. -3. Installiertes Paket per Introspektion inspizieren, um die echten Klassen für - Trigger (Conditions/Actions), Locations, Forces etc. zu lernen. -4. **Keine erratenen Methodennamen** schreiben. +```caddy +sc-mcp.pixel-by-design.de { + reverse_proxy sc-mcp:8000 +} +``` -## Stack & Architektur +Voraussetzung: Caddy und der `sc-mcp`-Container hängen im selben Docker-Netzwerk (in +`docker-compose.yml` als externes Netzwerk `caddy` eingetragen — passe den Namen an +dein Setup an). -| Aspekt | Wahl | -|---------------|------| -| Sprache | Python 3.12 | -| MCP-Framework | FastMCP (offizielles `mcp`-SDK, `pip install "mcp[cli]"`) | -| Transport | stdio (lokaler Test) + **Streamable HTTP** (Dauerbetrieb VPS) | -| Betrieb | Docker-Container, eingebunden ins bestehende docker-compose Setup | -| Reverse-Proxy | Caddy (läuft bereits) → TLS-Subdomain, z.B. `sc-mcp.pixel-by-design.de` | -| Karten-Volume | gemountetes Volume (z.B. `/data/maps`) für Basis-Karten, WAVs, Missionen | +### 3. In Claude als Custom Connector eintragen -## Tools (Prefix `sc_`) +Die URL, die du in Claude einträgst: -| Tool | Zweck | Hints | -|------|-------|-------| -| `sc_list_maps` | listet `.scm`/`.scx`-Dateien im Karten-Verzeichnis | readOnly | -| `sc_describe_map` | menschenlesbare Übersicht: Tileset, Größe, Forces/Player-Setup, Rassen, Locations, Trigger-Zusammenfassung | readOnly | -| `sc_list_locations` | Locations (MRGN) auflisten | readOnly | -| `sc_create_location` | Location anlegen | — | -| `sc_rename_location` | Location umbenennen | — | -| `sc_set_player_setup` | Forces (forc), Owner (ownr), Rassen (side) setzen | — | -| `sc_add_trigger` | **Kern-Tool.** Trigger mit Conditions + Actions + betroffenen Playern hinzufügen | — | -| `sc_list_triggers` | Trigger auflisten | readOnly | -| `sc_remove_trigger` | einzelnen Trigger entfernen | destructive | -| `sc_clear_triggers` | alle Trigger entfernen | destructive | -| `sc_embed_wav` | WAV-Datei einbetten und referenzierbar machen (wav-Editor) | — | -| `sc_save_map` | bearbeitetes Modell als neue `.scm`/`.scx` schreiben (`save_chk_to_mpq`, Basis-Karte als Vorlage) | — | -| `sc_set_briefing` | **optional / experimentell.** Mission-Briefing (MBRF). Falls kein High-Level-Editor: als TODO markieren, kein Blocker | — | +``` +https://sc-mcp.pixel-by-design.de/mcp +``` -### `sc_add_trigger` — Mindest-Abdeckung +In Claude (Web/Desktop): **Einstellungen → Connectors → Custom Connector hinzufügen** → +obige URL als Streamable-HTTP-/Remote-MCP-Endpunkt eintragen. Danach stehen die +`sc_`-Tools im Chat zur Verfügung. -Strukturierte Eingabe (Liste von Conditions, Liste von Actions, betroffene Player). -Muss u.a. abdecken: -- Einheiten erzeugen an Location -- Text einblenden (Display Text Message) -- Transmission mit Portrait + WAV -- Play WAV -- Sieg-/Niederlagebedingungen -- Timer -- AI-Skript ausführen +--- -### Anforderungen pro Tool +## Wo liegen Karten, WAVs und Missionen? -- Klare Beschreibung. -- Sinnvolle **Pydantic**-Eingabeschemas mit Feldbeschreibungen. -- Hilfreiche Fehlermeldungen, z.B. - `"Location 'Basis' existiert nicht – verfügbar: …"`. -- `readOnlyHint`/`destructiveHint`-Annotationen passend gesetzt. +Alles im Verzeichnis **`./data/maps`** (im Container als `/data/maps` gemountet): -## Definition of Done — Selbsttest +- **Basis-Karten** (`.scx`/`.scm`) — die Leinwand für neue Missionen. +- **WAV-Dateien** (`.wav`) — Voiceover/Sounds, die eingebettet werden. +- **Fertige Missionen** — schreibt der Server hierhin. -Ein kurzer Selbsttest beweist die ganze Pipeline: +Eine **Beispiel-Basis-Karte** liegt bereits unter +[`data/maps/base-map.scx`](./data/maps/base-map.scx) (256×256, Jungle), damit du sofort +die erste Mission bauen kannst. -1. Basis-Karte aus `/data/maps` laden. -2. Eine Location anlegen. -3. Per Trigger hinzufügen: bei Spielstart eine Texteinblendung - `"Mission 1 – Start"` **und** das Erzeugen von ein paar Einheiten an der Location. -4. Als neue `.scx` speichern. -5. Neue Datei erneut laden und bestätigen, dass Trigger + Location wirklich drin sind. +--- -Läuft das durch, steht das Fundament. +## Die Tools (Prefix `sc_`) -## Liefergegenstände +| Tool | Zweck | +|------|-------| +| `sc_list_maps` | listet alle Karten/Missionen im Verzeichnis *(nur lesen)* | +| `sc_describe_map` | Übersicht: Tileset, Größe, Player-Setup, Locations, Trigger-Zusammenfassung *(nur lesen)* | +| `sc_list_locations` | Locations auflisten *(nur lesen)* | +| `sc_create_location` | Location anlegen (Mittelpunkt + Größe in Pixeln) | +| `sc_rename_location` | Location umbenennen | +| `sc_set_player_setup` | Owner-Typ, Rasse und Force je Spieler setzen (+ Force-Namen) | +| `sc_add_trigger` | **Kern-Tool:** Trigger mit Bedingungen + Aktionen hinzufügen | +| `sc_list_triggers` | Trigger mit Klartext-Zusammenfassung auflisten *(nur lesen)* | +| `sc_remove_trigger` | einzelnen Trigger entfernen *(destruktiv)* | +| `sc_clear_triggers` | alle Trigger entfernen *(destruktiv)* | +| `sc_embed_wav` | WAV einbetten und referenzierbar machen | +| `sc_save_map` | aktuellen Stand als neue `.scx`/`.scm` schreiben | +| `sc_reset_map` | nicht gespeicherte Änderungen verwerfen *(destruktiv)* | -- [ ] Server läuft als Docker-Service, erreichbar über die TLS-Subdomain. -- [ ] **Ein** Start-/Neustart-Befehl. -- [ ] Die Connector-URL für Claude (Custom Connector). -- [ ] Beispiel-Basis-Karte liegt bereits in `/data/maps`. -- [ ] `README.md` mit Startbefehl, Subdomain-URL und Anleitung für den Custom - Connector in Claude. +**Arbeitsmodell:** Die Bearbeitung läuft pro Basis-Karte als Sitzung im Speicher des +Servers. Mehrere Tool-Aufrufe (Location anlegen, Trigger hinzufügen, WAV einbetten) +sammeln sich an; erst `sc_save_map` schreibt eine neue Datei. Die Basis-Karte bleibt +unverändert. `sc_reset_map` verwirft den Zwischenstand. -## Defaults / Entscheidungen +### Das Kern-Tool `sc_add_trigger` -> Hier werden während der Umsetzung die getroffenen Default-Entscheidungen -> dokumentiert, damit am Ende klar ist, was gewählt wurde (keine technischen -> Rückfragen, wo selbst entscheidbar). +Ein Trigger besteht aus **Bedingungen** (`conditions`, UND-verknüpft) und **Aktionen** +(`actions`), plus den **Spielern** (`players`), für die er läuft. Werte dürfen tolerant +angegeben werden: per Bezeichner (`TERRAN_MARINE`), Anzeigename (`Terran Marine`) oder +Zahl (`0`). -- _(wird ergänzt)_ +**Unterstützte Bedingungen (`type`):** `always`, `never`, `elapsed_time`, +`countdown_timer`, `bring`, `command`, `kill`, `deaths`, `accumulate`, `opponents`. -## Offene TODOs +**Unterstützte Aktionen (`type`):** `display_text`, `set_mission_objectives`, +`play_wav`, `create_unit`, `kill_unit_at_location`, `remove_unit_at_location`, +`move_unit`, `give_units`, `set_resources`, `set_deaths`, `set_countdown_timer`, +`pause_timer`, `unpause_timer`, `run_ai_script`, `run_ai_script_at_location`, +`center_view`, `minimap_ping`, `set_alliance_status`, `victory`, `defeat`. -- [ ] `richchk` installieren & per Introspektion echte Klassen verifizieren. -- [ ] Projektstruktur (FastMCP-Server, Pydantic-Schemas, Tool-Module) anlegen. -- [ ] Self-Test-Skript schreiben. -- [ ] Dockerfile + docker-compose-Service + Caddy-Reverse-Proxy-Eintrag. -- [ ] Beispiel-Basis-Karte bereitstellen. -- [ ] `sc_set_briefing` (MBRF) als best-effort prüfen. +Beispiel (Spielstart: Text + 4 Marines an der Location "Basis"): + +```json +{ + "map": "base-map.scx", + "players": ["PLAYER_1"], + "conditions": [{ "type": "always" }], + "actions": [ + { "type": "display_text", "text": "Mission 1 – Start" }, + { "type": "create_unit", "player": "PLAYER_1", "amount": 4, + "unit": "TERRAN_MARINE", "location": "Basis" } + ] +} +``` + +**Funksprüche / Voiceover:** Zuerst `sc_embed_wav` aufrufen (gibt einen `wav_path` +zurück), dann im Trigger `play_wav` (mit diesem `wav_path`) zusammen mit `display_text` +und optional `center_view` verwenden. + +--- + +## Selbsttest (beweist die ganze Pipeline) + +```bash +./run.sh selftest # im Container +# oder lokal: +python selftest.py +``` + +Der Test: Basis-Karte laden → Location anlegen → Trigger (Text + Einheiten bei Start) → +als neue `.scx` speichern → neu laden → Trigger + Location bestätigt. + +--- + +## Lokal testen (ohne Docker, stdio) + +```bash +python3 -m venv .venv && . .venv/bin/activate +pip install -r requirements.txt +SC_MAPS_DIR=./data/maps python -m starcraft_mcp.server # stdio-Transport +``` + +Für HTTP lokal: `SC_TRANSPORT=http SC_PORT=8000 SC_MAPS_DIR=./data/maps python -m starcraft_mcp.server` +→ Endpunkt `http://127.0.0.1:8000/mcp`. + +### Konfiguration (Umgebungsvariablen) + +| Variable | Standard | Bedeutung | +|----------|----------|-----------| +| `SC_TRANSPORT` | `stdio` | `http` für Streamable-HTTP-Betrieb | +| `SC_HOST` | `0.0.0.0` (HTTP) | Bind-Adresse | +| `SC_PORT` | `8000` | Port | +| `SC_MAPS_DIR` | `/data/maps` | Karten-/WAV-/Missions-Verzeichnis | + +--- + +## Gewählte Defaults + +- **Beispiel-Basis-Karte:** die MIT-lizenzierte `base-map.scx` aus dem RichChk-Repo + (256×256, Jungle). Liegt in `data/maps/`. +- **Locations** werden als Box um einen Mittelpunkt angelegt (Standard 96×96 px; + 32 px = 1 Kachel). +- **`preserve=true`** ist Standard bei `sc_add_trigger` (Trigger bleibt nach Auslösen + bestehen — passt für fast alle Kampagnen-Trigger). +- **RichChk-Logging** ist auf `CRITICAL` gesetzt, damit die Server-Ausgabe sauber bleibt. +- **Netzwerk:** Der Container wird **nicht** direkt nach außen exponiert; Caddy spricht + ihn intern über `sc-mcp:8000` an (sicherer). Zum reinen Lokaltest sind die `ports:` + in der `docker-compose.yml` auskommentiert vorbereitet. + +--- + +## Bekannte Einschränkungen / TODO + +- **Transmission mit Portrait (`Transmission`/`TalkingPortrait`):** RichChk hat in dieser + Version **kein** High-Level-Modell für diese Aktion. Funksprüche baust du deshalb aus + `play_wav` + `display_text` (+ optional `center_view`) zusammen. → TODO, sobald RichChk + es unterstützt. +- **Mission-Briefing (`sc_set_briefing`, MBRF):** RichChk bietet (noch) keinen + High-Level-Editor für die MBRF-Sektion. Für v1 bewusst weggelassen. → TODO. +- **Schalter (Switches) in Bedingungen/Aktionen:** für v1 nicht exponiert. → TODO. + +--- + +## Projektstruktur + +``` +starcraft_mcp/ + server.py FastMCP-Server mit allen sc_-Tools + workspace.py Karten-Sitzung im Speicher (Laden/Bearbeiten/Speichern, WAV-Einbettung) + triggers.py Pydantic-Schemas + Builder für Conditions/Actions (Herzstück-Doku) + enums.py tolerante Resolver (Einheiten, Spieler, Vergleiche, …) + richchk_logging.yaml setzt RichChk-Logging auf CRITICAL +selftest.py End-to-End-Selbsttest durch die echten Tools +data/maps/ Karten-Volume (Basis-Karte, WAVs, fertige Missionen) +Dockerfile, docker-compose.yml, Caddyfile, run.sh Betrieb +``` + +--- + +## Hinweis zur Erstellung + +Code und gesamte Pipeline wurden in einer Linux-Umgebung gebaut und **getestet** +(RichChk-Lese-/Schreib-Pipeline, alle 13 Tools, WAV-Einbettung, sowie der +Streamable-HTTP-Endpunkt per echtem MCP-Client-Handshake). Das tatsächliche Deployment +auf deinem Hetzner-VPS (Docker-Build, Caddy-Subdomain, DNS) führst du mit `./run.sh` +selbst aus — das konnte aus der Build-Umgebung heraus nicht für deinen VPS erfolgen. diff --git a/data/maps/.gitkeep b/data/maps/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/data/maps/base-map.scx b/data/maps/base-map.scx new file mode 100755 index 0000000..5cef423 Binary files /dev/null and b/data/maps/base-map.scx differ diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..08a21d2 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,37 @@ +# StarCraft-Kampagnen-MCP-Server ("Missions-Baumeister"). +# +# Einbindung in ein bestehendes Setup mit Caddy als Reverse-Proxy: +# - Der Server lauscht intern auf Port 8000 (Streamable HTTP, Pfad /mcp). +# - Caddy terminiert TLS und leitet die Subdomain an diesen Service weiter. +# Siehe Caddyfile in diesem Repo. +# +# Start/Neustart: ./run.sh (oder: docker compose up -d --build) + +services: + sc-mcp: + build: . + image: starcraft-campaign-mcp:latest + container_name: sc-mcp + restart: unless-stopped + environment: + SC_TRANSPORT: "http" + SC_HOST: "0.0.0.0" + SC_PORT: "8000" + SC_MAPS_DIR: "/data/maps" + volumes: + # Karten, WAVs und fertige Missionen liegen auf dem Host und bleiben erhalten. + - ./data/maps:/data/maps + # Standardmaessig NICHT nach aussen exponiert; Caddy spricht den Service im + # gemeinsamen Docker-Netzwerk ueber den Namen "sc-mcp:8000" an. + # Zum lokalen Testen ohne Caddy diese Zeilen einkommentieren: + # ports: + # - "8000:8000" + networks: + - web + +# Externes Netzwerk, in dem auch Caddy haengt. Passe den Namen an dein Setup an +# (z.B. das von deinem bestehenden Caddy genutzte Netzwerk). +networks: + web: + external: true + name: caddy diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..eb7b33c --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +# Laufzeit-Abhaengigkeiten des StarCraft-Kampagnen-MCP-Servers. +richchk==0.1.1 +mcp[cli]==1.28.0 +PyYAML>=6.0 diff --git a/run.sh b/run.sh new file mode 100755 index 0000000..19a9173 --- /dev/null +++ b/run.sh @@ -0,0 +1,40 @@ +#!/usr/bin/env bash +# Ein-Befehl-Start/Neustart des StarCraft-Kampagnen-MCP-Servers. +# +# ./run.sh -> baut (falls noetig) und startet/neustartet den Container +# ./run.sh logs -> zeigt die Live-Logs +# ./run.sh selftest -> fuehrt den Selbsttest im Container aus +# ./run.sh stop -> stoppt den Container +set -euo pipefail +cd "$(dirname "$0")" + +# docker compose (v2) bevorzugen, sonst docker-compose (v1). +if docker compose version >/dev/null 2>&1; then + DC="docker compose" +else + DC="docker-compose" +fi + +cmd="${1:-up}" +case "$cmd" in + up|"") + $DC up -d --build + echo + echo "Server laeuft. Streamable-HTTP-Endpunkt intern: http://sc-mcp:8000/mcp" + echo "Oeffentlich (via Caddy): https://sc-mcp.pixel-by-design.de/mcp" + ;; + logs) + $DC logs -f sc-mcp + ;; + selftest) + $DC run --rm sc-mcp python selftest.py + ;; + stop|down) + $DC down + ;; + *) + echo "Unbekannter Befehl: $cmd" + echo "Nutze: ./run.sh [up|logs|selftest|stop]" + exit 1 + ;; +esac diff --git a/selftest.py b/selftest.py new file mode 100755 index 0000000..e06ec83 --- /dev/null +++ b/selftest.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +"""Selbsttest: beweist die komplette Pipeline durch die echten MCP-Tools. + +Ablauf (Definition of Done): + 1. Basis-Karte aus dem Karten-Verzeichnis laden. + 2. Eine Location anlegen. + 3. Trigger hinzufuegen: bei Spielstart Text "Mission 1 - Start" einblenden UND + ein paar Einheiten an der Location erzeugen. + 4. Als neue .scx speichern. + 5. Neue Datei erneut laden und bestaetigen, dass Trigger + Location drin sind. + +Start: python selftest.py [basis-karte.scx] +Exit-Code 0 = bestanden. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import shutil +import sys +import tempfile + + +def _data(result) -> dict: + """Hole das strukturierte Ergebnis aus einem FastMCP call_tool-Resultat.""" + # FastMCP gibt (content_blocks, structured_result) zurueck. + if isinstance(result, tuple) and len(result) == 2: + structured = result[1] + if isinstance(structured, dict): + return structured + # Fallback: erstes TextContent als JSON parsen. + blocks = result[0] if isinstance(result, tuple) else result + for b in blocks: + text = getattr(b, "text", None) + if text: + try: + return json.loads(text) + except json.JSONDecodeError: + return {"text": text} + return {} + + +async def run(base_map: str) -> bool: + workdir = tempfile.mkdtemp(prefix="sc_selftest_") + os.environ["SC_MAPS_DIR"] = workdir + shutil.copyfile(base_map, os.path.join(workdir, os.path.basename(base_map))) + base_name = os.path.basename(base_map) + + # Workspace-Modul liest SC_MAPS_DIR beim Import -> erst jetzt importieren. + from starcraft_mcp import workspace # noqa: E402 + from starcraft_mcp.server import mcp # noqa: E402 + + workspace.MAPS_DIR = workdir # falls Modul bereits importiert war + + print(f"[1] Basis-Karte: {base_name} (Verzeichnis: {workdir})") + res = _data(await mcp.call_tool("sc_list_maps", {})) + assert base_name in res["maps"], f"Basis-Karte nicht gelistet: {res}" + + print("[2] Location 'Basis' anlegen ...") + res = _data( + await mcp.call_tool( + "sc_create_location", + {"map": base_name, "name": "Basis", "center_x": 512, "center_y": 512}, + ) + ) + assert res.get("ok"), f"create_location fehlgeschlagen: {res}" + print(f" -> Index {res['index']}, Box {res['box']}") + + print("[3] Trigger hinzufuegen (Text + 4 Marines bei Spielstart) ...") + res = _data( + await mcp.call_tool( + "sc_add_trigger", + { + "map": base_name, + "players": ["PLAYER_1"], + "conditions": [{"type": "always"}], + "actions": [ + {"type": "display_text", "text": "Mission 1 - Start"}, + { + "type": "create_unit", + "player": "PLAYER_1", + "amount": 4, + "unit": "TERRAN_MARINE", + "location": "Basis", + }, + ], + }, + ) + ) + assert res.get("ok"), f"add_trigger fehlgeschlagen: {res}" + print(f" -> {res['summary']}") + + print("[4] Als 'mission1.scx' speichern ...") + res = _data( + await mcp.call_tool( + "sc_save_map", + {"map": base_name, "output_name": "mission1.scx", "overwrite": True}, + ) + ) + assert res.get("ok"), f"save_map fehlgeschlagen: {res}" + out_path = res["path"] + print(f" -> {out_path} ({os.path.getsize(out_path)} Bytes)") + + print("[5] Neue Datei unabhaengig neu laden und verifizieren ...") + from richchk.io.mpq.starcraft_mpq_io_helper import StarCraftMpqIoHelper + from richchk.io.richchk.query.chk_query_util import ChkQueryUtil + from richchk.model.richchk.mrgn.rich_mrgn_section import RichMrgnSection + from richchk.model.richchk.trig.rich_trig_section import RichTrigSection + + chk = StarCraftMpqIoHelper.create_mpq_io().read_chk_from_mpq(out_path) + mrgn = ChkQueryUtil.find_only_rich_section_in_chk(RichMrgnSection, chk) + trig = ChkQueryUtil.find_only_rich_section_in_chk(RichTrigSection, chk) + loc_names = [l._custom_location_name.value for l in mrgn.locations] + assert "Basis" in loc_names, f"Location 'Basis' fehlt nach Reload: {loc_names}" + assert len(trig.triggers) >= 1, "Kein Trigger nach Reload gefunden" + action_types = [type(a).__name__ for a in trig.triggers[-1].actions] + assert "DisplayTextMessageAction" in action_types, action_types + assert "CreateUnitAction" in action_types, action_types + print(" -> Location 'Basis' vorhanden: True") + print(f" -> Trigger-Anzahl: {len(trig.triggers)}") + print(" -> Aktionen im Trigger: DisplayText + CreateUnit bestaetigt") + + shutil.rmtree(workdir, ignore_errors=True) + return True + + +def _find_default_base_map() -> str: + here = os.path.dirname(os.path.abspath(__file__)) + candidate = os.path.join(here, "data", "maps", "base-map.scx") + return candidate + + +def main() -> int: + base_map = sys.argv[1] if len(sys.argv) > 1 else _find_default_base_map() + if not os.path.exists(base_map): + print(f"FEHLER: Basis-Karte nicht gefunden: {base_map}") + return 2 + try: + ok = asyncio.run(run(base_map)) + except AssertionError as exc: + print(f"\nSELBSTTEST FEHLGESCHLAGEN: {exc}") + return 1 + except Exception as exc: # noqa: BLE001 + print(f"\nSELBSTTEST FEHLER: {type(exc).__name__}: {exc}") + return 1 + if ok: + print("\n=== SELBSTTEST BESTANDEN ===") + return 0 + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/starcraft_mcp/__init__.py b/starcraft_mcp/__init__.py new file mode 100644 index 0000000..301821c --- /dev/null +++ b/starcraft_mcp/__init__.py @@ -0,0 +1,21 @@ +"""StarCraft-Kampagnen-MCP-Server ("Missions-Baumeister"). + +Ein MCP-Server, der StarCraft-Brood-War-Karten (.scm/.scx) liest und schreibt, +damit Claude aus einer Kreativ-Vorlage echte, spielbare Missionsdateien baut. +Die Karten-Manipulation laeuft komplett ueber die Bibliothek RichChk. +""" + +import os as _os + +# RichChk loggt sonst sehr ausfuehrlich (INFO/WARNING) nach stderr. Wir setzen das +# Log-Level per RichChk-Config-Datei auf CRITICAL, BEVOR irgendein RichChk-Logger +# erzeugt wird (RichChk liest die Datei beim ersten get_logger). Das haelt die Ausgabe +# sauber; unsere Tools melden Fehler ohnehin als Exceptions. +_os.environ.setdefault( + "io.sethmachine.richchk.config", + _os.path.join(_os.path.dirname(__file__), "richchk_logging.yaml"), +) + +__all__ = ["__version__"] + +__version__ = "0.1.0" diff --git a/starcraft_mcp/enums.py b/starcraft_mcp/enums.py new file mode 100644 index 0000000..344418f --- /dev/null +++ b/starcraft_mcp/enums.py @@ -0,0 +1,111 @@ +"""Flexible Resolver fuer RichChk-Enums. + +Claude (und Menschen) sollen Einheiten, Spieler, Vergleiche usw. tolerant angeben +koennen: per Bezeichner ("TERRAN_MARINE"), per Anzeigename ("Terran Marine") oder per +Zahl ("0"). Diese Helfer uebersetzen solche Eingaben in die echten RichChk-Enum-Member +und werfen bei Fehlern eine *hilfreiche* Fehlermeldung mit Beispielen. +""" + +from __future__ import annotations + +from typing import Type, TypeVar + +from richchk.model.richchk.forc.force_id import ForceId +from richchk.model.richchk.ownr.player_type import PlayerType +from richchk.model.richchk.richchk_enum import RichChkEnum +from richchk.model.richchk.side.player_race import PlayerRace +from richchk.model.richchk.trig.conditions.comparators.numeric_comparator import ( + NumericComparator, +) +from richchk.model.richchk.trig.enums.alliance_status import AllianceStatus +from richchk.model.richchk.trig.enums.amount_modifier import AmountModifier +from richchk.model.richchk.trig.enums.resource_type import ResourceType +from richchk.model.richchk.trig.enums.switch_state import SwitchState +from richchk.model.richchk.trig.player_id import PlayerId +from richchk.model.richchk.unis.unit_id import UnitId + +_E = TypeVar("_E", bound=RichChkEnum) + + +def _identifier(member: RichChkEnum) -> str: + """Der Python-Bezeichner eines Members, z.B. 'TERRAN_MARINE'.""" + return member._name_ # type: ignore[attr-defined] + + +def _build_index(enum_cls: Type[_E]) -> dict[str, _E]: + index: dict[str, _E] = {} + for member in enum_cls: # type: ignore[attr-defined] + index[_identifier(member).upper()] = member + index[member.name.upper()] = member # Anzeigename ("Terran Marine") + index[str(member.id)] = member # numerische ID + return index + + +def _examples(enum_cls: Type[_E], limit: int = 20) -> str: + names = sorted({_identifier(m) for m in enum_cls}) # type: ignore[attr-defined] + shown = ", ".join(names[:limit]) + if len(names) > limit: + shown += f", … ({len(names)} insgesamt)" + return shown + + +def resolve(enum_cls: Type[_E], value: str | int, what: str) -> _E: + """Loese eine tolerante Eingabe in ein RichChk-Enum-Member auf. + + Akzeptiert Bezeichner ('TERRAN_MARINE'), Anzeigename ('Terran Marine') oder ID (0). + """ + index = _build_index(enum_cls) + key = str(value).strip().upper() + if key in index: + return index[key] + raise ValueError( + f"Unbekannter Wert fuer {what}: {value!r}. " + f"Verfuegbar (Bezeichner): {_examples(enum_cls)}" + ) + + +# --- Convenience-Wrapper mit sprechenden Namen -------------------------------- + + +def resolve_unit(value: str | int) -> UnitId: + return resolve(UnitId, value, "Einheit (unit)") + + +def resolve_player(value: str | int) -> PlayerId: + return resolve(PlayerId, value, "Spieler/Gruppe (player)") + + +def resolve_comparator(value: str | int) -> NumericComparator: + return resolve(NumericComparator, value, "Vergleich (comparator)") + + +def resolve_modifier(value: str | int) -> AmountModifier: + return resolve(AmountModifier, value, "Mengen-Modifikator (modifier)") + + +def resolve_resource(value: str | int) -> ResourceType: + return resolve(ResourceType, value, "Ressource (resource)") + + +def resolve_alliance(value: str | int) -> AllianceStatus: + return resolve(AllianceStatus, value, "Allianz-Status (alliance_status)") + + +def resolve_switch_state(value: str | int) -> SwitchState: + return resolve(SwitchState, value, "Schalter-Zustand (switch_state)") + + +def resolve_player_type(value: str | int) -> PlayerType: + return resolve(PlayerType, value, "Spieler-Typ (type)") + + +def resolve_race(value: str | int) -> PlayerRace: + return resolve(PlayerRace, value, "Rasse (race)") + + +def resolve_force(value: str | int) -> ForceId: + return resolve(ForceId, value, "Force/Team (force)") + + +def list_identifiers(enum_cls: Type[_E]) -> list[str]: + return sorted({_identifier(m) for m in enum_cls}) # type: ignore[attr-defined] diff --git a/starcraft_mcp/richchk_logging.yaml b/starcraft_mcp/richchk_logging.yaml new file mode 100644 index 0000000..0e19f0f --- /dev/null +++ b/starcraft_mcp/richchk_logging.yaml @@ -0,0 +1,2 @@ +logging: + level: CRITICAL diff --git a/starcraft_mcp/server.py b/starcraft_mcp/server.py new file mode 100644 index 0000000..0ca0fe9 --- /dev/null +++ b/starcraft_mcp/server.py @@ -0,0 +1,515 @@ +"""FastMCP-Server: StarCraft-Kampagnen-MCP ("Missions-Baumeister"). + +Stellt `sc_`-Tools bereit, mit denen Claude StarCraft-Brood-War-Karten (.scm/.scx) +liest und schreibt. Die Karten-Manipulation laeuft komplett ueber RichChk. + +Start: + - stdio (lokaler Test): python -m starcraft_mcp.server + - Streamable HTTP (Betrieb): SC_TRANSPORT=http python -m starcraft_mcp.server +""" + +from __future__ import annotations + +import os +from typing import Optional + +from mcp.server.fastmcp import FastMCP +from mcp.types import ToolAnnotations +from richchk.editor.richchk.rich_forc_editor import RichForcEditor +from richchk.editor.richchk.rich_ownr_editor import RichOwnrEditor +from richchk.editor.richchk.rich_side_editor import RichSideEditor +from richchk.editor.richchk.rich_trig_editor import RichTrigEditor +from richchk.model.richchk.dim.rich_dim_section import RichDimSection +from richchk.model.richchk.era.rich_era_section import RichEraSection +from richchk.model.richchk.forc.rich_forc_section import RichForcSection +from richchk.model.richchk.ownr.rich_ownr_section import RichOwnrSection +from richchk.model.richchk.side.rich_side_section import RichSideSection +from richchk.model.richchk.trig.actions.preserve_trigger_action import PreserveTrigger +from richchk.model.richchk.trig.rich_trigger import RichTrigger +from pydantic import Field + +from . import enums, triggers, workspace +from .triggers import Action, Condition + +mcp = FastMCP( + "starcraft-campaign-mcp", + instructions=( + "Baue spielbare StarCraft-Brood-War-Missionen aus einer Basis-Karte. " + "Das Gelaende wird NICHT generiert - jede Mission startet von einer " + "vorhandenen Basis-Karte in /data/maps; nur Logik (Trigger), Texte, " + "Locations, Player-Setup und Sounds werden bearbeitet. Typischer Ablauf: " + "sc_list_maps -> sc_describe_map -> sc_create_location -> sc_add_trigger " + "-> sc_embed_wav -> sc_save_map." + ), +) + +_READONLY = ToolAnnotations(readOnlyHint=True, destructiveHint=False) +_EDIT = ToolAnnotations(readOnlyHint=False, destructiveHint=False) +_DESTRUCTIVE = ToolAnnotations(readOnlyHint=False, destructiveHint=True) + + +# --- Lese-Tools --------------------------------------------------------------- + + +@mcp.tool( + annotations=_READONLY, + description="Listet alle Basis-Karten und fertigen Missionen (.scm/.scx) im " + "Karten-Verzeichnis.", +) +def sc_list_maps() -> dict: + maps = workspace.list_maps() + return { + "maps_dir": workspace.MAPS_DIR, + "count": len(maps), + "maps": maps, + } + + +@mcp.tool( + annotations=_READONLY, + description="Liest eine Karte und gibt eine menschenlesbare Uebersicht: Tileset, " + "Groesse, Player-Setup (Typ/Rasse/Force), vorhandene Locations sowie Anzahl und " + "Klartext-Zusammenfassung der Trigger. Spiegelt den aktuellen Bearbeitungsstand, " + "falls die Karte bereits geoeffnet ist.", +) +def sc_describe_map( + map: str = Field(description="Dateiname der Karte in /data/maps."), +) -> dict: + ws = workspace.open_workspace(map) + dim = ws.section(RichDimSection) + era = ws.section(RichEraSection) + ownr = ws.section(RichOwnrSection) + side = ws.section(RichSideSection) + forc = ws.section(RichForcSection) + + players = [] + for i in range(8): + players.append( + { + "player": f"PLAYER_{i + 1}", + "type": ownr.player_types[i].name + if i < len(ownr.player_types) + else "?", + "race": side.player_races[i].name + if i < len(side.player_races) + else "?", + "force": forc.player_force_assignments[i].name + if i < len(forc.player_force_assignments) + else "?", + } + ) + + locations = [ + { + "name": loc._custom_location_name.value or f"(unbenannt #{loc.index})", + "index": loc.index, + "box": [loc._left_x1, loc._top_y1, loc._right_x2, loc._bottom_y2], + } + for loc in ws.mrgn.locations + ] + + trigger_summaries = [] + for i, tr in enumerate(ws.trig.triggers): + trigger_summaries.append(_summarize_trigger(i, tr)) + + return { + "map": ws.base_name, + "tileset": era.tileset.name, + "size": {"width": dim.width, "height": dim.height}, + "players": players, + "forces": [ + {"force": f"FORCE_{i + 1}", "name": f.name.value or f"Force {i + 1}"} + for i, f in enumerate(forc.forces) + ], + "locations": locations, + "trigger_count": len(ws.trig.triggers), + "triggers": trigger_summaries, + "pending_wav_embeds": [os.path.basename(w[0]) for w in ws.pending_wavs], + } + + +@mcp.tool( + annotations=_READONLY, + description="Listet alle Locations (MRGN) der Karte mit Index und Bounding-Box.", +) +def sc_list_locations( + map: str = Field(description="Dateiname der Karte in /data/maps."), +) -> dict: + ws = workspace.open_workspace(map) + return { + "map": ws.base_name, + "locations": [ + { + "name": loc._custom_location_name.value or f"(unbenannt #{loc.index})", + "index": loc.index, + "box": [loc._left_x1, loc._top_y1, loc._right_x2, loc._bottom_y2], + } + for loc in ws.mrgn.locations + ], + } + + +@mcp.tool( + annotations=_READONLY, + description="Listet alle Trigger der geoeffneten Karte mit Index, betroffenen " + "Spielern und einer Klartext-Zusammenfassung der Bedingungen und Aktionen.", +) +def sc_list_triggers( + map: str = Field(description="Dateiname der Karte in /data/maps."), +) -> dict: + ws = workspace.open_workspace(map) + return { + "map": ws.base_name, + "trigger_count": len(ws.trig.triggers), + "triggers": [_summarize_trigger(i, t) for i, t in enumerate(ws.trig.triggers)], + } + + +# --- Locations ---------------------------------------------------------------- + + +@mcp.tool( + annotations=_EDIT, + description="Legt eine neue Location (MRGN) an. Koordinaten in Pixeln (32 px = " + "1 Kachel). Es wird eine Box um den Mittelpunkt gebaut. Locations werden in " + "Triggern ueber ihren Namen referenziert.", +) +def sc_create_location( + map: str = Field(description="Dateiname der Karte in /data/maps."), + name: str = Field(description="Eindeutiger Name der Location, z.B. 'Basis'."), + center_x: int = Field(description="Mittelpunkt X in Pixeln."), + center_y: int = Field(description="Mittelpunkt Y in Pixeln."), + width: int = Field(default=96, description="Breite der Box in Pixeln (Standard 96)."), + height: int = Field( + default=96, description="Hoehe der Box in Pixeln (Standard 96)." + ), +) -> dict: + ws = workspace.open_workspace(map) + loc = ws.add_location(name, center_x, center_y, width, height) + return { + "ok": True, + "name": name, + "index": loc.index, + "box": [loc._left_x1, loc._top_y1, loc._right_x2, loc._bottom_y2], + "hinweis": "Aenderung ist im Speicher. Mit sc_save_map persistieren.", + } + + +@mcp.tool( + annotations=_EDIT, + description="Benennt eine existierende Location um.", +) +def sc_rename_location( + map: str = Field(description="Dateiname der Karte in /data/maps."), + old_name: str = Field(description="Aktueller Name der Location."), + new_name: str = Field(description="Neuer Name der Location."), +) -> dict: + ws = workspace.open_workspace(map) + loc = ws.rename_location(old_name, new_name) + return {"ok": True, "name": new_name, "index": loc.index} + + +# --- Player-Setup ------------------------------------------------------------- + + +@mcp.tool( + annotations=_EDIT, + description=( + "Setzt das Player-Setup: Owner-Typ (OWNR), Rasse (SIDE) und Force-Zuordnung " + "(FORC) je Spieler. Optional Force-Namen.\n" + "type: INACTIVE, COMPUTER_GAME, HUMAN_OCCUPIED, RESCUE_PASSIVE, COMPUTER, " + "HUMAN, NEUTRAL, CLOSED.\n" + "race: ZERG, TERRAN, PROTOSS, INDEPENDENT, NEUTRAL, USER_SELECT, RANDOM, " + "INACTIVE.\n" + "force: FORCE_1..FORCE_4." + ), +) +def sc_set_player_setup( + map: str = Field(description="Dateiname der Karte in /data/maps."), + players: list[dict] = Field( + description="Liste von Eintraegen je Spieler, z.B. " + '[{"player":"PLAYER_1","type":"HUMAN","race":"TERRAN","force":"FORCE_1"}]. ' + "Die Felder type/race/force sind jeweils optional." + ), + force_names: Optional[list[dict]] = Field( + default=None, + description='Optionale Force-Namen, z.B. [{"force":"FORCE_1","name":"Allianz"}].', + ), +) -> dict: + ws = workspace.open_workspace(map) + applied = [] + for entry in players: + player = enums.resolve_player(entry["player"]) + if entry.get("type") is not None: + ownr = ws.section(RichOwnrSection) + ws.replace( + RichOwnrEditor().set_player_type( + player, enums.resolve_player_type(entry["type"]), ownr + ) + ) + if entry.get("race") is not None: + side = ws.section(RichSideSection) + ws.replace( + RichSideEditor().set_player_race( + player, enums.resolve_race(entry["race"]), side + ) + ) + if entry.get("force") is not None: + forc = ws.section(RichForcSection) + ws.replace( + RichForcEditor().add_player_to_force( + player, enums.resolve_force(entry["force"]), forc + ) + ) + applied.append(entry["player"]) + + if force_names: + from richchk.model.richchk.forc.rich_force import RichForce + from richchk.model.richchk.str.rich_string import RichString + + for fn in force_names: + force = enums.resolve_force(fn["force"]) + forc = ws.section(RichForcSection) + existing = forc.forces[force.id] + ws.replace( + RichForcEditor().set_force( + force, + RichForce(_name=RichString(fn["name"]), _flags=existing.flags), + forc, + ) + ) + + return { + "ok": True, + "updated_players": applied, + "updated_forces": [f["force"] for f in force_names] if force_names else [], + "hinweis": "Aenderung ist im Speicher. Mit sc_save_map persistieren.", + } + + +# --- Trigger (Kern-Tool) ------------------------------------------------------ + + +@mcp.tool( + annotations=_EDIT, + description=triggers.__doc__ + + "\n\nFuegt EINEN Trigger hinzu. `players` ist die Liste der Spieler/Gruppen, " + "fuer die der Trigger laeuft (z.B. ['PLAYER_1'] oder ['ALL_PLAYERS']). Mit " + "`preserve=true` (Standard) bleibt der Trigger nach dem Ausloesen bestehen " + "(empfohlen fuer fast alle Kampagnen-Trigger). Locations werden ueber ihren " + "Namen referenziert und muessen vorher existieren.", +) +def sc_add_trigger( + map: str = Field(description="Dateiname der Karte in /data/maps."), + players: list[str] = Field( + description="Spieler/Gruppen, fuer die der Trigger laeuft, z.B. " + "['PLAYER_1'], ['ALL_PLAYERS'], ['FORCE_1']." + ), + conditions: list[Condition] = Field( + description="Liste der Bedingungen (UND-verknuepft). Leere Liste = 'always'." + ), + actions: list[Action] = Field(description="Liste der Aktionen in Reihenfolge."), + preserve: bool = Field( + default=True, + description="Trigger nach Ausloesen erhalten (Preserve Trigger). Standard true.", + ), +) -> dict: + ws = workspace.open_workspace(map) + + if not players: + raise ValueError("Mindestens ein Spieler/eine Gruppe in 'players' angeben.") + player_set = {enums.resolve_player(p) for p in players} + + rich_conditions = [ + triggers.build_condition(c, ws.resolve_location) for c in conditions + ] or [_always()] + rich_actions = [triggers.build_action(a, ws.resolve_location) for a in actions] + if preserve: + rich_actions.append(PreserveTrigger()) + if not rich_actions: + raise ValueError("Ein Trigger braucht mindestens eine Aktion.") + + trigger = RichTrigger( + _conditions=rich_conditions, + _actions=rich_actions, + _players=player_set, + ) + new_trig = RichTrigEditor.add_triggers([trigger], ws.trig) + ws.replace(new_trig) + + return { + "ok": True, + "trigger_index": len(ws.trig.triggers) - 1, + "summary": _summarize_trigger(len(ws.trig.triggers) - 1, trigger), + "hinweis": "Aenderung ist im Speicher. Mit sc_save_map persistieren.", + } + + +@mcp.tool( + annotations=_DESTRUCTIVE, + description="Entfernt EINEN Trigger anhand seines Index (siehe sc_list_triggers).", +) +def sc_remove_trigger( + map: str = Field(description="Dateiname der Karte in /data/maps."), + index: int = Field(description="0-basierter Index des Triggers."), +) -> dict: + from richchk.model.richchk.trig.rich_trig_section import RichTrigSection + + ws = workspace.open_workspace(map) + current = list(ws.trig.triggers) + if not 0 <= index < len(current): + raise ValueError( + f"Trigger-Index {index} ungueltig. Es gibt {len(current)} Trigger (0..{len(current) - 1})." + ) + removed = current.pop(index) + ws.replace(RichTrigSection(_triggers=current)) + return { + "ok": True, + "removed_index": index, + "removed_summary": _summarize_trigger(index, removed), + "remaining": len(current), + } + + +@mcp.tool( + annotations=_DESTRUCTIVE, + description="Entfernt ALLE Trigger der Karte. Vorsicht: nicht umkehrbar (bis zum " + "erneuten Laden der Basis-Karte).", +) +def sc_clear_triggers( + map: str = Field(description="Dateiname der Karte in /data/maps."), +) -> dict: + from richchk.model.richchk.trig.rich_trig_section import RichTrigSection + + ws = workspace.open_workspace(map) + count = len(ws.trig.triggers) + ws.replace(RichTrigSection(_triggers=[])) + return {"ok": True, "removed": count} + + +# --- Sounds ------------------------------------------------------------------- + + +@mcp.tool( + annotations=_EDIT, + description="Bettet eine WAV-Datei (Voiceover/Sound) in die Karte ein und macht sie " + "referenzierbar. Die WAV muss als Datei im Karten-Verzeichnis liegen. Die " + "Einbettung passiert beim Speichern (sc_save_map). Rueckgabe ist der in-MPQ-Pfad, " + "den du in play_wav-Aktionen (sc_add_trigger) als `wav_path` angibst.", +) +def sc_embed_wav( + map: str = Field(description="Dateiname der Karte in /data/maps."), + wav_filename: str = Field( + description="Dateiname der WAV im Karten-Verzeichnis, z.B. 'funk1.wav'." + ), +) -> dict: + ws = workspace.open_workspace(map) + wav_path = workspace.map_path(wav_filename) + if not os.path.exists(wav_path): + available = [ + f + for f in (os.listdir(workspace.MAPS_DIR) if os.path.isdir(workspace.MAPS_DIR) else []) + if f.lower().endswith(".wav") + ] + raise FileNotFoundError( + f"WAV {os.path.basename(wav_filename)!r} nicht gefunden in " + f"{workspace.MAPS_DIR}. Verfuegbar: {', '.join(available) or '(keine)'}." + ) + in_mpq = ws.embed_wav(wav_path) + return { + "ok": True, + "wav_path": in_mpq, + "hinweis": "Nutze diesen wav_path in play_wav-Aktionen. Die Datei wird beim " + "sc_save_map physisch in die Karte eingebettet.", + } + + +# --- Speichern ---------------------------------------------------------------- + + +@mcp.tool( + annotations=_EDIT, + description="Schreibt den aktuellen Bearbeitungsstand als neue Missionsdatei " + "(.scm/.scx) ins Karten-Verzeichnis. Die Basis-Karte dient als Vorlage und bleibt " + "unveraendert. Eingebettete WAVs werden hier hinzugefuegt.", +) +def sc_save_map( + map: str = Field(description="Dateiname der Basis-Karte in /data/maps."), + output_name: str = Field( + description="Dateiname der Ausgabe, z.B. 'mission1.scx'. Ohne Endung wird " + ".scx ergaenzt." + ), + overwrite: bool = Field( + default=False, description="Bestehende Datei ueberschreiben." + ), +) -> dict: + ws = workspace.open_workspace(map) + out_path = ws.save_as(output_name, overwrite) + return { + "ok": True, + "output": os.path.basename(out_path), + "path": out_path, + "embedded_wavs": [os.path.basename(w[0]) for w in ws.pending_wavs], + "triggers": len(ws.trig.triggers), + "locations": len(ws.mrgn.locations), + } + + +@mcp.tool( + annotations=_DESTRUCTIVE, + description="Verwirft alle nicht gespeicherten Aenderungen und laedt die Basis-" + "Karte frisch von der Platte.", +) +def sc_reset_map( + map: str = Field(description="Dateiname der Karte in /data/maps."), +) -> dict: + ws = workspace.open_workspace(map, fresh=True) + return { + "ok": True, + "map": ws.base_name, + "triggers": len(ws.trig.triggers), + "locations": len(ws.mrgn.locations), + } + + +# --- Helfer ------------------------------------------------------------------- + + +def _always(): + from richchk.model.richchk.trig.conditions.always_condition import AlwaysCondition + + return AlwaysCondition() + + +def _summarize_trigger(index: int, tr: RichTrigger) -> dict: + players = sorted(p.name for p in tr.players) + conds = [ + triggers.summarize_condition(c) + for c in tr.conditions + if type(c).__name__ not in ("NoConditionCondition",) + ] + acts = [ + triggers.summarize_action(a) + for a in tr.actions + if type(a).__name__ not in ("NoActionAction",) + ] + return { + "index": index, + "players": players, + "conditions": conds, + "actions": acts, + } + + +def main() -> None: + transport = os.environ.get("SC_TRANSPORT", "stdio").lower() + if transport in ("http", "streamable-http", "streamable_http"): + mcp.settings.host = os.environ.get("SC_HOST", "0.0.0.0") + mcp.settings.port = int(os.environ.get("SC_PORT", "8000")) + mcp.run(transport="streamable-http") + else: + mcp.run() + + +if __name__ == "__main__": + main() diff --git a/starcraft_mcp/triggers.py b/starcraft_mcp/triggers.py new file mode 100644 index 0000000..eaf91dd --- /dev/null +++ b/starcraft_mcp/triggers.py @@ -0,0 +1,452 @@ +"""Bausteine fuer Trigger: Pydantic-Eingaben -> echte RichChk-Objekte. + +Das ist das Herzstueck. Ein Trigger besteht aus *Bedingungen* (conditions) und +*Aktionen* (actions). Hier werden die tolerant getippten Eingaben von Claude in die +echten RichChk-Modelle uebersetzt, mit hilfreichen Fehlermeldungen. + +Unterstuetzte CONDITION-Typen (Feld `type`): + - always : immer wahr (Standard-Bedingung) + - never : nie wahr + - elapsed_time : seit Spielstart sind N Sekunden vergangen + Felder: seconds, comparator (AT_LEAST|AT_MOST|EXACTLY) + - countdown_timer : der Countdown-Timer vergleicht mit N Sekunden + Felder: seconds, comparator + - bring : Spieler hat N Einheiten an einer Location + Felder: player, comparator, amount, unit, location + - command : Spieler kommandiert N Einheiten (ganze Karte) + Felder: player, comparator, amount, unit + - kill : Spieler hat N Einheiten eines Typs getoetet + Felder: player, comparator, amount, unit + - deaths : Spieler hat N Todesfaelle eines Einheitentyps + Felder: player, comparator, amount, unit + - accumulate : Spieler hat N Ressourcen + Felder: player, comparator, amount, resource (ORE|GAS|ORE_AND_GAS) + - opponents : Spieler hat N verbleibende Gegner + Felder: player, comparator, amount + +Unterstuetzte ACTION-Typen (Feld `type`): + - display_text : Text einblenden. Felder: text + - set_mission_objectives : Missionsziele setzen. Felder: text + - play_wav : WAV abspielen. Felder: wav_path (in-MPQ-Pfad), duration_ms? + - create_unit : Einheiten erzeugen. Felder: player, amount, unit, location + - kill_unit_at_location : Einheiten toeten. Felder: player, amount, unit, location + - remove_unit_at_location: Einheiten entfernen. Felder: player, amount, unit, location + - move_unit : Einheiten bewegen. + Felder: player, amount, unit, location (Quelle), destination (Ziel) + - give_units : Einheiten uebergeben. + Felder: player (von), target_player (an), amount, unit, location + - set_resources : Ressourcen setzen. + Felder: player, modifier (SET_TO|ADD|SUBTRACT), amount, resource + - set_deaths : Todeszaehler setzen. + Felder: player, modifier, amount, unit + - set_countdown_timer : Countdown-Timer setzen. Felder: seconds, modifier + - pause_timer / unpause_timer : Countdown-Timer pausieren/fortsetzen + - run_ai_script : AI-Skript ausfuehren. Felder: ai_script (4-Zeichen-Code) + - run_ai_script_at_location : AI-Skript an Location. Felder: ai_script, location + - center_view : Kamera zentrieren. Felder: location + - minimap_ping : Minimap-Ping. Felder: location + - set_alliance_status : Allianz setzen. + Felder: player (Ziel-Gruppe), alliance_status (ENEMY|ALLY|ALLIED_VICTORY) + - victory : Mission gewonnen + - defeat : Mission verloren + +Hinweis "Transmission mit Portrait": RichChk hat (Stand dieser Version) kein +High-Level-Modell fuer die Transmission-/TalkingPortrait-Aktion. Baue einen +Funkspruch stattdessen aus `play_wav` + `display_text` (+ optional `center_view`) +zusammen. Siehe README. +""" + +from __future__ import annotations + +from typing import Callable, Optional + +from pydantic import BaseModel, Field +from richchk.model.richchk.mrgn.rich_location import RichLocation +from richchk.model.richchk.str.rich_string import RichString +from richchk.model.richchk.trig.actions.center_view_action import CenterViewAction +from richchk.model.richchk.trig.actions.create_unit_action import CreateUnitAction +from richchk.model.richchk.trig.actions.defeat_action import DefeatAction +from richchk.model.richchk.trig.actions.display_text_message_action import ( + DisplayTextMessageAction, +) +from richchk.model.richchk.trig.actions.give_unit_action import GiveUnitAction +from richchk.model.richchk.trig.actions.kill_unit_at_location_action import ( + KillUnitAtLocationAction, +) +from richchk.model.richchk.trig.actions.minimap_ping_action import MinimapPingAction +from richchk.model.richchk.trig.actions.move_unit_action import MoveUnitAction +from richchk.model.richchk.trig.actions.pause_countdown_timer import ( + PauseCountdownTimerAction, +) +from richchk.model.richchk.trig.actions.play_wav_action import PlayWavAction +from richchk.model.richchk.trig.actions.remove_unit_at_location_action import ( + RemoveUnitAtLocationAction, +) +from richchk.model.richchk.trig.actions.run_ai_script_action import RunAiScriptAction +from richchk.model.richchk.trig.actions.run_ai_script_at_location_action import ( + RunAiScriptAtLocationAction, +) +from richchk.model.richchk.trig.actions.set_countdown_timer import ( + SetCountdownTimerAction, +) +from richchk.model.richchk.trig.actions.set_alliance_status_action import ( + SetAllianceStatusAction, +) +from richchk.model.richchk.trig.actions.set_deaths_action import SetDeathsAction +from richchk.model.richchk.trig.actions.set_mission_objectives_action import ( + SetMissionObjectivesAction, +) +from richchk.model.richchk.trig.actions.set_resources_action import SetResourcesAction +from richchk.model.richchk.trig.actions.unpause_countdown_timer import ( + UnpauseCountdownTimerAction, +) +from richchk.model.richchk.trig.actions.victory_action import VictoryAction +from richchk.model.richchk.trig.conditions.accumulate_resources_condition import ( + AccumulateResourcesCondition, +) +from richchk.model.richchk.trig.conditions.always_condition import AlwaysCondition +from richchk.model.richchk.trig.conditions.bring_condition import BringCondition +from richchk.model.richchk.trig.conditions.command_condition import CommandCondition +from richchk.model.richchk.trig.conditions.countdown_timer_condition import ( + CountdownTimerCondition, +) +from richchk.model.richchk.trig.conditions.deaths_condition import DeathsCondition +from richchk.model.richchk.trig.conditions.elapsed_time_condition import ( + ElapsedTimeCondition, +) +from richchk.model.richchk.trig.conditions.kill_condition import KillCondition +from richchk.model.richchk.trig.conditions.never_condition import NeverCondition +from richchk.model.richchk.trig.conditions.opponents_remaining_condition import ( + OpponentsRemainingCondition, +) +from richchk.model.richchk.trig.enums.ai_script import AiScript, KnownAiScript + +from . import enums + +# Locations werden ueber einen Resolver (Name -> RichLocation) aufgeloest, den der +# Aufrufer injiziert, weil die verfuegbaren Locations vom Workspace abhaengen. +LocationResolver = Callable[[str], RichLocation] + + +class Condition(BaseModel): + """Eine einzelne Trigger-Bedingung. Pflichtfelder haengen von `type` ab.""" + + type: str = Field( + description="Bedingungstyp: always, never, elapsed_time, countdown_timer, " + "bring, command, kill, deaths, accumulate, opponents." + ) + player: Optional[str] = Field( + default=None, + description="Spieler/Gruppe, z.B. PLAYER_1, ALL_PLAYERS, FOES, CURRENT_PLAYER.", + ) + comparator: str = Field( + default="AT_LEAST", + description="Vergleich: AT_LEAST, AT_MOST oder EXACTLY.", + ) + amount: Optional[int] = Field(default=None, description="Menge/Anzahl/Ressourcen.") + unit: Optional[str] = Field( + default=None, description="Einheit, z.B. TERRAN_MARINE oder 'Terran Marine'." + ) + location: Optional[str] = Field( + default=None, description="Name einer existierenden Location." + ) + seconds: Optional[int] = Field(default=None, description="Sekunden (Zeit-Typen).") + resource: Optional[str] = Field( + default=None, description="Ressource: ORE, GAS oder ORE_AND_GAS." + ) + + +class Action(BaseModel): + """Eine einzelne Trigger-Aktion. Pflichtfelder haengen von `type` ab.""" + + type: str = Field( + description="Aktionstyp: display_text, set_mission_objectives, play_wav, " + "create_unit, kill_unit_at_location, remove_unit_at_location, move_unit, " + "give_units, set_resources, set_deaths, set_countdown_timer, pause_timer, " + "unpause_timer, run_ai_script, run_ai_script_at_location, center_view, " + "minimap_ping, set_alliance_status, victory, defeat." + ) + text: Optional[str] = Field(default=None, description="Anzuzeigender Text.") + wav_path: Optional[str] = Field( + default=None, + description="In-MPQ-Pfad der WAV (Rueckgabe von sc_embed_wav), z.B. " + "'staredit\\\\wav\\\\funk1.wav'.", + ) + duration_ms: Optional[int] = Field( + default=None, description="Dauer der WAV in Millisekunden (optional)." + ) + player: Optional[str] = Field( + default=None, description="Betroffener Spieler/Gruppe (Quelle bei give_units)." + ) + target_player: Optional[str] = Field( + default=None, description="Zielspieler bei give_units." + ) + amount: int = Field( + default=1, description="Anzahl Einheiten / Menge. 0 bedeutet 'Alle'." + ) + unit: Optional[str] = Field(default=None, description="Einheit, z.B. TERRAN_MARINE.") + location: Optional[str] = Field( + default=None, description="Name einer existierenden Location." + ) + destination: Optional[str] = Field( + default=None, description="Ziel-Location (bei move_unit)." + ) + seconds: Optional[int] = Field(default=None, description="Sekunden (Timer).") + modifier: str = Field( + default="SET_TO", description="Mengen-Modifikator: SET_TO, ADD oder SUBTRACT." + ) + resource: Optional[str] = Field( + default=None, description="Ressource: ORE, GAS oder ORE_AND_GAS." + ) + ai_script: Optional[str] = Field( + default=None, + description="AI-Skript: 4-Zeichen-Code (z.B. 'TMCx') oder bekannter Name " + "(z.B. JUNKYARD_DOG).", + ) + alliance_status: Optional[str] = Field( + default=None, description="Allianz: ENEMY, ALLY oder ALLIED_VICTORY." + ) + + +def _require(value, field: str, action_or_cond_type: str): + if value is None: + raise ValueError( + f"Feld '{field}' fehlt fuer Typ '{action_or_cond_type}'. " + f"Bitte '{field}' angeben." + ) + return value + + +def _resolve_ai_script(value: str) -> AiScript: + """Bekannter Name (JUNKYARD_DOG) oder roher 4-Zeichen-Code.""" + key = value.strip() + try: + return KnownAiScript[key.upper()].value + except KeyError: + pass + if len(key) == 4: + return AiScript(key, key) + known = ", ".join(sorted(k.name for k in KnownAiScript)) + raise ValueError( + f"Unbekanntes AI-Skript {value!r}. Gib einen 4-Zeichen-Code an " + f"(z.B. 'TMCx') oder einen bekannten Namen: {known}" + ) + + +def build_condition(cond: Condition, resolve_location: LocationResolver): + """Uebersetze eine Condition-Eingabe in ein RichChk-Conditionobjekt.""" + t = cond.type.strip().lower() + if t == "always": + return AlwaysCondition() + if t == "never": + return NeverCondition() + if t == "elapsed_time": + return ElapsedTimeCondition( + _seconds=_require(cond.seconds, "seconds", t), + _comparator=enums.resolve_comparator(cond.comparator), + ) + if t == "countdown_timer": + return CountdownTimerCondition( + _seconds=_require(cond.seconds, "seconds", t), + _comparator=enums.resolve_comparator(cond.comparator), + ) + if t == "bring": + return BringCondition( + _group=enums.resolve_player(_require(cond.player, "player", t)), + _comparator=enums.resolve_comparator(cond.comparator), + _amount=_require(cond.amount, "amount", t), + _unit=enums.resolve_unit(_require(cond.unit, "unit", t)), + _location=resolve_location(_require(cond.location, "location", t)), + ) + if t == "command": + return CommandCondition( + _group=enums.resolve_player(_require(cond.player, "player", t)), + _comparator=enums.resolve_comparator(cond.comparator), + _amount=_require(cond.amount, "amount", t), + _unit=enums.resolve_unit(_require(cond.unit, "unit", t)), + ) + if t == "kill": + return KillCondition( + _group=enums.resolve_player(_require(cond.player, "player", t)), + _comparator=enums.resolve_comparator(cond.comparator), + _amount=_require(cond.amount, "amount", t), + _unit=enums.resolve_unit(_require(cond.unit, "unit", t)), + ) + if t == "deaths": + return DeathsCondition( + _group=enums.resolve_player(_require(cond.player, "player", t)), + _comparator=enums.resolve_comparator(cond.comparator), + _amount=_require(cond.amount, "amount", t), + _unit=enums.resolve_unit(_require(cond.unit, "unit", t)), + ) + if t == "accumulate": + return AccumulateResourcesCondition( + _group=enums.resolve_player(_require(cond.player, "player", t)), + _comparator=enums.resolve_comparator(cond.comparator), + _amount=_require(cond.amount, "amount", t), + _resource=enums.resolve_resource(_require(cond.resource, "resource", t)), + ) + if t == "opponents": + return OpponentsRemainingCondition( + _group=enums.resolve_player(_require(cond.player, "player", t)), + _amount=_require(cond.amount, "amount", t), + _comparator=enums.resolve_comparator(cond.comparator), + ) + raise ValueError( + f"Unbekannter Bedingungstyp: {cond.type!r}. Erlaubt: always, never, " + "elapsed_time, countdown_timer, bring, command, kill, deaths, accumulate, " + "opponents." + ) + + +def build_action(act: Action, resolve_location: LocationResolver): + """Uebersetze eine Action-Eingabe in ein RichChk-Actionobjekt.""" + t = act.type.strip().lower() + if t == "display_text": + return DisplayTextMessageAction(RichString(_require(act.text, "text", t))) + if t == "set_mission_objectives": + return SetMissionObjectivesAction(RichString(_require(act.text, "text", t))) + if t == "play_wav": + return PlayWavAction( + _path_to_wav_in_mpq=_require(act.wav_path, "wav_path", t), + _duration_ms=act.duration_ms, + ) + if t == "create_unit": + return CreateUnitAction( + enums.resolve_player(_require(act.player, "player", t)), + act.amount, + enums.resolve_unit(_require(act.unit, "unit", t)), + resolve_location(_require(act.location, "location", t)), + ) + if t == "kill_unit_at_location": + return KillUnitAtLocationAction( + enums.resolve_player(_require(act.player, "player", t)), + act.amount, + enums.resolve_unit(_require(act.unit, "unit", t)), + resolve_location(_require(act.location, "location", t)), + ) + if t == "remove_unit_at_location": + return RemoveUnitAtLocationAction( + enums.resolve_player(_require(act.player, "player", t)), + act.amount, + enums.resolve_unit(_require(act.unit, "unit", t)), + resolve_location(_require(act.location, "location", t)), + ) + if t == "move_unit": + return MoveUnitAction( + enums.resolve_unit(_require(act.unit, "unit", t)), + enums.resolve_player(_require(act.player, "player", t)), + act.amount, + resolve_location(_require(act.location, "location", t)), + resolve_location(_require(act.destination, "destination", t)), + ) + if t == "give_units": + return GiveUnitAction( + enums.resolve_player(_require(act.player, "player", t)), + enums.resolve_player(_require(act.target_player, "target_player", t)), + act.amount, + enums.resolve_unit(_require(act.unit, "unit", t)), + resolve_location(_require(act.location, "location", t)), + ) + if t == "set_resources": + return SetResourcesAction( + enums.resolve_player(_require(act.player, "player", t)), + enums.resolve_modifier(act.modifier), + _require(act.amount, "amount", t), + enums.resolve_resource(_require(act.resource, "resource", t)), + ) + if t == "set_deaths": + return SetDeathsAction( + enums.resolve_player(_require(act.player, "player", t)), + enums.resolve_unit(_require(act.unit, "unit", t)), + _require(act.amount, "amount", t), + enums.resolve_modifier(act.modifier), + ) + if t == "set_countdown_timer": + return SetCountdownTimerAction( + _require(act.seconds, "seconds", t), + enums.resolve_modifier(act.modifier), + ) + if t == "pause_timer": + return PauseCountdownTimerAction() + if t == "unpause_timer": + return UnpauseCountdownTimerAction() + if t == "run_ai_script": + return RunAiScriptAction(_resolve_ai_script(_require(act.ai_script, "ai_script", t))) + if t == "run_ai_script_at_location": + return RunAiScriptAtLocationAction( + _resolve_ai_script(_require(act.ai_script, "ai_script", t)), + resolve_location(_require(act.location, "location", t)), + ) + if t == "center_view": + return CenterViewAction(resolve_location(_require(act.location, "location", t))) + if t == "minimap_ping": + return MinimapPingAction(resolve_location(_require(act.location, "location", t))) + if t == "set_alliance_status": + return SetAllianceStatusAction( + enums.resolve_player(_require(act.player, "player", t)), + enums.resolve_alliance(_require(act.alliance_status, "alliance_status", t)), + ) + if t == "victory": + return VictoryAction() + if t == "defeat": + return DefeatAction() + raise ValueError( + f"Unbekannter Aktionstyp: {act.type!r}. Siehe Tool-Beschreibung fuer die Liste." + ) + + +# --- Klartext-Zusammenfassung fuer describe_map / list_triggers ---------------- + + +def summarize_condition(cond) -> str: + name = type(cond).__name__.replace("Condition", "") + parts = [name] + for attr, label in ( + ("group", "player"), + ("comparator", "cmp"), + ("amount", "n"), + ("unit", "unit"), + ("seconds", "s"), + ("resource", "res"), + ("location", "loc"), + ): + if hasattr(cond, attr): + val = getattr(cond, attr) + parts.append(f"{label}={_pretty(val)}") + return " ".join(parts) + + +def summarize_action(act) -> str: + name = type(act).__name__.replace("Action", "") + parts = [name] + for attr, label in ( + ("group", "player"), + ("from_group", "from"), + ("to_group", "to"), + ("amount", "n"), + ("unit", "unit"), + ("location", "loc"), + ("destination_location", "dst"), + ("message", "text"), + ("path_to_wav_in_mpq", "wav"), + ("seconds", "s"), + ("resource", "res"), + ("alliance_status", "ally"), + ): + if hasattr(act, attr): + parts.append(f"{label}={_pretty(getattr(act, attr))}") + return " ".join(parts) + + +def _pretty(val) -> str: + if isinstance(val, RichString): + return repr(val.value) + if isinstance(val, RichLocation): + return repr(val._custom_location_name.value or f"#{val.index}") + if hasattr(val, "name"): + try: + return str(val.name) + except Exception: + pass + return str(val) diff --git a/starcraft_mcp/workspace.py b/starcraft_mcp/workspace.py new file mode 100644 index 0000000..da5bd58 --- /dev/null +++ b/starcraft_mcp/workspace.py @@ -0,0 +1,256 @@ +"""Arbeits-Session pro Basis-Karte. + +MCP-Tool-Aufrufe sind einzeln, aber das Bauen einer Mission besteht aus vielen +Schritten (Location anlegen, mehrere Trigger, WAV einbetten, speichern). Diese Klasse +haelt den bearbeiteten RichChk-Zustand einer Karte im Speicher zusammen mit einer +Namens-Registry fuer Locations und den noch einzubettenden WAV-Dateien. + +Workspaces leben prozessweit (Modul-Singleton-Dict), passend zum Streamable-HTTP- +Server, der ein einzelner Prozess ist. +""" + +from __future__ import annotations + +import os +import threading +from dataclasses import dataclass, field + +from richchk.editor.richchk.rich_chk_editor import RichChkEditor +from richchk.editor.richchk.rich_mrgn_editor import RichMrgnEditor +from richchk.io.mpq.starcraft_mpq_io_helper import StarCraftMpqIoHelper +from richchk.io.richchk.query.chk_query_util import ChkQueryUtil +from richchk.model.richchk.mrgn.rich_location import RichLocation +from richchk.model.richchk.mrgn.rich_mrgn_section import RichMrgnSection +from richchk.model.richchk.rich_chk import RichChk +from richchk.model.richchk.str.rich_string import RichString +from richchk.model.richchk.trig.rich_trig_section import RichTrigSection + +# Verzeichnis mit Basis-Karten, WAVs und fertigen Missionen (gemountetes Volume). +MAPS_DIR = os.environ.get("SC_MAPS_DIR", "/data/maps") + +_MPQ_IO = None +_MPQ_LOCK = threading.Lock() + + +def get_mpq_io(): + """Lazy-Singleton fuer den (relativ teuren) StormLib-Loader.""" + global _MPQ_IO + with _MPQ_LOCK: + if _MPQ_IO is None: + _MPQ_IO = StarCraftMpqIoHelper.create_mpq_io() + return _MPQ_IO + + +def map_path(name: str) -> str: + """Absoluter Pfad einer Karte/WAV im Karten-Verzeichnis (kein Ausbrechen).""" + base = os.path.basename(name) + return os.path.join(MAPS_DIR, base) + + +@dataclass +class Workspace: + """Bearbeitbarer Zustand genau einer Basis-Karte.""" + + base_name: str + base_path: str + chk: RichChk + locations: dict[str, RichLocation] = field(default_factory=dict) + # noch physisch einzubettende WAVs: (Disk-Pfad, In-MPQ-Pfad) + pending_wavs: list[tuple[str, str]] = field(default_factory=list) + + # --- Sektion-Helfer ------------------------------------------------------- + + def section(self, section_type): + return ChkQueryUtil.find_only_rich_section_in_chk(section_type, self.chk) + + def replace(self, new_section) -> None: + self.chk = RichChkEditor().replace_chk_section(new_section, self.chk) + + @property + def mrgn(self) -> RichMrgnSection: + return self.section(RichMrgnSection) + + @property + def trig(self) -> RichTrigSection: + return self.section(RichTrigSection) + + # --- Locations ------------------------------------------------------------ + + def reload_location_registry(self) -> None: + self.locations = {} + for loc in self.mrgn.locations: + name = loc._custom_location_name.value + if name: + self.locations[name] = loc + + def resolve_location(self, name: str) -> RichLocation: + if name in self.locations: + return self.locations[name] + available = ", ".join(sorted(self.locations)) or "(keine)" + raise ValueError( + f"Location {name!r} existiert nicht. Verfuegbar: {available}. " + f"Lege sie zuerst mit sc_create_location an." + ) + + def add_location( + self, + name: str, + center_x: int, + center_y: int, + width: int, + height: int, + ) -> RichLocation: + if name in self.locations: + raise ValueError( + f"Location {name!r} existiert bereits. Nutze sc_rename_location oder " + f"einen anderen Namen." + ) + half_w, half_h = max(1, width // 2), max(1, height // 2) + loc = RichLocation( + _left_x1=max(0, center_x - half_w), + _top_y1=max(0, center_y - half_h), + _right_x2=center_x + half_w, + _bottom_y2=center_y + half_h, + _custom_location_name=RichString(name), + ) + new_mrgn, _ = RichMrgnEditor().add_locations([loc], self.mrgn) + self.replace(new_mrgn) + # Das allokierte Objekt (mit Index) aus der neuen Sektion ziehen. + allocated = next( + l for l in new_mrgn.locations if l._custom_location_name.value == name + ) + self.locations[name] = allocated + return allocated + + def rename_location(self, old_name: str, new_name: str) -> RichLocation: + if old_name not in self.locations: + available = ", ".join(sorted(self.locations)) or "(keine)" + raise ValueError( + f"Location {old_name!r} existiert nicht. Verfuegbar: {available}." + ) + if new_name in self.locations: + raise ValueError(f"Location {new_name!r} existiert bereits.") + old = self.locations[old_name] + renamed = RichLocation( + _left_x1=old._left_x1, + _top_y1=old._top_y1, + _right_x2=old._right_x2, + _bottom_y2=old._bottom_y2, + _custom_location_name=RichString(new_name), + _index=old._index, + _low_elevation=old._low_elevation, + _medium_elevation=old._medium_elevation, + _high_elevation=old._high_elevation, + _low_air=old._low_air, + _medium_air=old._medium_air, + _high_air=old._high_air, + ) + new_locations = [ + renamed if l._index == old._index else l for l in self.mrgn.locations + ] + self.replace(RichMrgnSection(_locations=new_locations)) + del self.locations[old_name] + self.locations[new_name] = renamed + return renamed + + # --- Sounds --------------------------------------------------------------- + + def embed_wav(self, wav_disk_path: str) -> str: + """Registriere eine WAV im WAV-Section (damit der String beim Encode bekannt + ist) und merke die Datei zur physischen Einbettung beim Speichern vor. + + :return: der In-MPQ-Pfad, der in play_wav-Aktionen referenziert wird. + """ + from richchk.editor.richchk.rich_wav_editor import RichWavEditor + from richchk.model.richchk.wav.rich_wav_section import RichWavSection + + in_mpq = "staredit\\wav\\" + os.path.basename(wav_disk_path) + wav_section = self.section(RichWavSection) + self.replace(RichWavEditor().add_wav_files([in_mpq], wav_section)) + if not any(p[1] == in_mpq for p in self.pending_wavs): + self.pending_wavs.append((wav_disk_path, in_mpq)) + return in_mpq + + # --- Speichern ------------------------------------------------------------ + + def save_as(self, output_name: str, overwrite: bool) -> str: + out_path = map_path(output_name) + if not (out_path.lower().endswith(".scx") or out_path.lower().endswith(".scm")): + out_path += ".scx" + if os.path.exists(out_path) and not overwrite: + raise FileExistsError( + f"Datei existiert bereits: {os.path.basename(out_path)}. " + f"Setze overwrite=true zum Ueberschreiben." + ) + mpq_io = get_mpq_io() + # CHK (inkl. aller Edits; die WAV-Eintraege wurden bei sc_embed_wav registriert) + # in die neue MPQ schreiben. + mpq_io.save_chk_to_mpq( + self.chk, self.base_path, out_path, overwrite_existing=True + ) + # Danach die physischen WAV-Dateien in die fertige MPQ kopieren. + if self.pending_wavs: + from richchk.model.mpq.stormlib.stormlib_archive_mode import ( + StormLibArchiveMode, + ) + + sw = mpq_io._stormlib_wrapper + handle = sw.open_archive(out_path, StormLibArchiveMode.STORMLIB_WRITE_ONLY) + try: + for disk_path, in_mpq in self.pending_wavs: + sw.add_file( + handle, + infile=disk_path, + path_to_file_in_archive=in_mpq, + overwrite_existing=True, + ) + sw.compact_archive(handle) + finally: + sw.close_archive(handle) + return out_path + + +# --- Workspace-Registry ------------------------------------------------------- + +_WORKSPACES: dict[str, Workspace] = {} +_WS_LOCK = threading.Lock() + + +def open_workspace(map_name: str, fresh: bool = False) -> Workspace: + """Lade-oder-hole den Workspace einer Basis-Karte. + + :param map_name: Dateiname der Basis-Karte in MAPS_DIR. + :param fresh: True erzwingt ein Neuladen von der Platte (verwirft Edits). + """ + base = os.path.basename(map_name) + with _WS_LOCK: + if not fresh and base in _WORKSPACES: + return _WORKSPACES[base] + path = map_path(base) + if not os.path.exists(path): + available = ", ".join(list_maps()) or "(keine)" + raise FileNotFoundError( + f"Karte {base!r} nicht gefunden in {MAPS_DIR}. Verfuegbar: {available}." + ) + chk = get_mpq_io().read_chk_from_mpq(path) + ws = Workspace(base_name=base, base_path=path, chk=chk) + ws.reload_location_registry() + with _WS_LOCK: + _WORKSPACES[base] = ws + return ws + + +def discard_workspace(map_name: str) -> bool: + base = os.path.basename(map_name) + with _WS_LOCK: + return _WORKSPACES.pop(base, None) is not None + + +def list_maps() -> list[str]: + if not os.path.isdir(MAPS_DIR): + return [] + return sorted( + f + for f in os.listdir(MAPS_DIR) + if f.lower().endswith((".scx", ".scm")) + )