StarCraft-Kampagnen-MCP-Server ("Missions-Baumeister") (#1)
* Projekt-Spezifikation für StarCraft-Kampagnen-MCP-Server in README festhalten Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NaZBZofC1on2gkSMb5UWZy * StarCraft-Kampagnen-MCP-Server: vollstaendige Implementierung FastMCP-Server (stdio + Streamable HTTP) mit 13 sc_-Tools fuer das Lesen und Schreiben von StarCraft-Brood-War-Karten (.scm/.scx) ueber RichChk: - sc_list_maps, sc_describe_map, sc_list_locations, sc_list_triggers (lesen) - sc_create_location, sc_rename_location, sc_set_player_setup - sc_add_trigger (Kern-Tool: 10 Condition- und 20 Action-Typen), sc_remove_trigger, sc_clear_triggers - sc_embed_wav (Voiceover/Sound-Einbettung), sc_save_map, sc_reset_map Tolerante Pydantic-Eingaben, hilfreiche Fehlermeldungen, readOnly/destructive-Hints. Karten-Sitzung im Speicher pro Basis-Karte; Basis-Karte bleibt unveraendert. Deployment: Dockerfile (python:3.12-slim), docker-compose, Caddyfile, run.sh. Beispiel-Basis-Karte in data/maps. Selbsttest beweist die ganze Pipeline. Verifiziert: Pipeline + alle Tools + WAV-Einbettung + Streamable-HTTP-Handshake (echter MCP-Client), unter Python 3.11 und 3.12. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01NaZBZofC1on2gkSMb5UWZy --------- Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Kenearos
GitHub
parent
57502546bc
commit
4075cf4ebd
17 changed files with 1886 additions and 1 deletions
21
starcraft_mcp/__init__.py
Normal file
21
starcraft_mcp/__init__.py
Normal file
|
|
@ -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"
|
||||
111
starcraft_mcp/enums.py
Normal file
111
starcraft_mcp/enums.py
Normal file
|
|
@ -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]
|
||||
2
starcraft_mcp/richchk_logging.yaml
Normal file
2
starcraft_mcp/richchk_logging.yaml
Normal file
|
|
@ -0,0 +1,2 @@
|
|||
logging:
|
||||
level: CRITICAL
|
||||
515
starcraft_mcp/server.py
Normal file
515
starcraft_mcp/server.py
Normal file
|
|
@ -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()
|
||||
452
starcraft_mcp/triggers.py
Normal file
452
starcraft_mcp/triggers.py
Normal file
|
|
@ -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)
|
||||
256
starcraft_mcp/workspace.py
Normal file
256
starcraft_mcp/workspace.py
Normal file
|
|
@ -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"))
|
||||
)
|
||||
Loading…
Add table
Add a link
Reference in a new issue