Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 KoalaTrapLord
|
3a94de4c1f | ||
|
KoalaTrapLord
|
c52323dd1d |
No files matched your search
-31
@@ -38,34 +38,3 @@ Thumbs.db
|
||||
|
||||
# Logs
|
||||
*.log
|
||||
|
||||
# Precon decks — runtime data, downloaded locally by Forge/scripts
|
||||
# Do not commit .dck files; they are large and machine-generated
|
||||
precon-decks/*.dck
|
||||
precon-decks/precon-index.json
|
||||
|
||||
# Phase 0: Local Scryfall card image cache (~1.5-2GB)
|
||||
# Downloaded by scripts/download_card_images.py — never commit
|
||||
static/card-images/
|
||||
|
||||
# Phase 0: SQLite image index (generated by scripts/build_image_index.py)
|
||||
data/card_image_index.db
|
||||
|
||||
# Phase 6: AI decision logs and training sets (large runtime data)
|
||||
data/decision_logs/
|
||||
data/game_results.jsonl
|
||||
data/training_sets/
|
||||
|
||||
# Unity generated / local config files
|
||||
unity-client/Packages/packages-lock.json
|
||||
unity-client/ProjectSettings/EditorBuildSettings.asset
|
||||
unity-client/ProjectSettings/EditorSettings.asset
|
||||
unity-client/ProjectSettings/ShaderGraphSettings.asset
|
||||
unity-commander-ui/Packages/packages-lock.json
|
||||
unity-commander-ui/ProjectSettings/EditorBuildSettings.asset
|
||||
unity-commander-ui/ProjectSettings/EditorSettings.asset
|
||||
|
||||
# Third-party Unity assets — installed locally via Unity Package Manager / Asset Store
|
||||
# These are large binary asset packs; each dev installs from their own license
|
||||
unity-client/Assets/Eric VFX Studio/
|
||||
unity-client/Assets/TcgEngine/
|
||||
@@ -1 +0,0 @@
|
||||
# Phase 2: Dual-Brain AI Decision Engine
|
||||
@@ -1,5 +0,0 @@
|
||||
# Phase 6 - decision_logger.py
|
||||
# Logs per-decision: game_id, turn, phase, player_id, personality,
|
||||
# brain_used, prompt, raw_response, parsed_action, fallback_used
|
||||
# Configurable via LOG_AI_DECISIONS env var
|
||||
# See issue #173
|
||||
@@ -1,15 +0,0 @@
|
||||
# Phase 2 - game_orchestrator.py *** CORE PRIORITY ***
|
||||
# GameAIOrchestrator: select_brain(), assign_personalities(), get_action()
|
||||
# Brains: DeepSeek-R1-14B (complex) + GPT-OSS-20B (tactical)
|
||||
# ChromaDB RAG for per-player card knowledge
|
||||
# See issue #169
|
||||
|
||||
PERSONALITY_POOL = [
|
||||
"aggressive -- attack early and often, prioritize players with lowest life totals",
|
||||
"control -- counter key spells, hold mana open, only act when clearly advantageous",
|
||||
"combo -- develop mana and card advantage quietly, avoid conflict until ready to win",
|
||||
"political -- make deals, attack whoever is ahead, never fully commit to one target",
|
||||
"pillowfort -- build defensive pieces, let others fight, win slowly through attrition",
|
||||
"stompy -- play the biggest threats as fast as possible, ignore table politics",
|
||||
"tempo -- disrupt opponents early, trade efficiently, stay ahead on board",
|
||||
]
|
||||
@@ -1,12 +0,0 @@
|
||||
# Phase 0: Card image cache and data enrichment
|
||||
from commander_ai_lab.cards.image_cache import get_image_cache, ImageCache, ImageCacheResult
|
||||
from commander_ai_lab.cards.card_data_enricher import enrich_card, enrich_card_list, enrich_game_state
|
||||
|
||||
__all__ = [
|
||||
"get_image_cache",
|
||||
"ImageCache",
|
||||
"ImageCacheResult",
|
||||
"enrich_card",
|
||||
"enrich_card_list",
|
||||
"enrich_game_state",
|
||||
]
|
||||
@@ -1,116 +0,0 @@
|
||||
"""
|
||||
Phase 0 — card_data_enricher.py
|
||||
|
||||
Injects art_url and art_url_back into card data dicts/DTOs
|
||||
before they are broadcast to Unity via WebSocket.
|
||||
|
||||
Used by:
|
||||
- Phase 1 game_state_parser.py (enriches cards in GameStateDTO)
|
||||
- Phase 3 WebSocket handler (enriches full game state before broadcast)
|
||||
|
||||
Lookup chain per card:
|
||||
1. scryfall_id present -> ImageCache.lookup_by_id()
|
||||
2. scryfall_id absent -> ImageCache.lookup_by_name() (Forge card name)
|
||||
3. Both fail -> Scryfall CDN fallback URL
|
||||
4. CDN fallback absent -> empty string (Unity shows card back texture)
|
||||
|
||||
See issue #167
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from commander_ai_lab.cards.image_cache import get_image_cache, ImageCacheResult
|
||||
|
||||
log = logging.getLogger("commander_ai_lab.cards.card_data_enricher")
|
||||
|
||||
# Scryfall CDN fallback when both local cache and DB are unavailable
|
||||
_CDN_FRONT = "https://cards.scryfall.io/normal/front/{d1}/{d2}/{sid}.jpg"
|
||||
_CDN_BACK = "https://cards.scryfall.io/normal/back/{d1}/{d2}/{sid}.jpg"
|
||||
|
||||
|
||||
def _cdn_fallback_urls(scryfall_id: str) -> tuple[str, str | None]:
|
||||
if not scryfall_id:
|
||||
return "", None
|
||||
d1 = scryfall_id[0]
|
||||
d2 = scryfall_id[1] if len(scryfall_id) > 1 else "0"
|
||||
front = _CDN_FRONT.format(d1=d1, d2=d2, sid=scryfall_id)
|
||||
back = _CDN_BACK.format(d1=d1, d2=d2, sid=scryfall_id)
|
||||
return front, back
|
||||
|
||||
|
||||
def enrich_card(card: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Inject art_url and art_url_back into a single card dict.
|
||||
Mutates in place AND returns the card for convenience.
|
||||
|
||||
Input card dict is expected to have at minimum:
|
||||
- 'name': str (card name from Forge)
|
||||
- 'scryfall_id': str (optional but preferred)
|
||||
|
||||
After enrichment the card will have:
|
||||
- 'art_url': str front face image URL
|
||||
- 'art_url_back': str|None back face image URL (DFCs only)
|
||||
- 'has_back': bool
|
||||
"""
|
||||
# Skip if already enriched
|
||||
if card.get("art_url"):
|
||||
return card
|
||||
|
||||
cache = get_image_cache()
|
||||
result: ImageCacheResult | None = None
|
||||
|
||||
scryfall_id = card.get("scryfall_id", "")
|
||||
if scryfall_id:
|
||||
result = cache.lookup_by_id(scryfall_id)
|
||||
|
||||
if result is None:
|
||||
name = card.get("name", "")
|
||||
if name:
|
||||
result = cache.lookup_by_name(name)
|
||||
|
||||
if result is not None:
|
||||
card["art_url"] = result.front_url
|
||||
card["art_url_back"] = result.back_url
|
||||
card["has_back"] = result.has_back
|
||||
else:
|
||||
# Hard fallback: build CDN URL from scryfall_id if available
|
||||
front, back = _cdn_fallback_urls(scryfall_id)
|
||||
card["art_url"] = front
|
||||
card["art_url_back"] = back if back else None
|
||||
card["has_back"] = False
|
||||
if not front:
|
||||
log.warning("No art URL resolved for card: %s (id=%s)", card.get("name"), scryfall_id)
|
||||
|
||||
return card
|
||||
|
||||
|
||||
def enrich_card_list(cards: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
"""Enrich a list of card dicts. Mutates in place."""
|
||||
for card in cards:
|
||||
enrich_card(card)
|
||||
return cards
|
||||
|
||||
|
||||
def enrich_game_state(state: dict[str, Any]) -> dict[str, Any]:
|
||||
"""
|
||||
Enrich all card lists in a full GameStateDTO dict.
|
||||
Walks all players and all zones: hand, battlefield, graveyard, exile, command_zone.
|
||||
Mutates in place AND returns state.
|
||||
"""
|
||||
zone_keys = ("hand", "battlefield", "graveyard", "exile", "command_zone")
|
||||
|
||||
for player in state.get("players", []):
|
||||
for zone in zone_keys:
|
||||
cards = player.get(zone)
|
||||
if isinstance(cards, list):
|
||||
enrich_card_list(cards)
|
||||
|
||||
# Also enrich the stack
|
||||
for stack_item in state.get("stack", []):
|
||||
card = stack_item.get("card")
|
||||
if isinstance(card, dict):
|
||||
enrich_card(card)
|
||||
|
||||
return state
|
||||
@@ -1,167 +0,0 @@
|
||||
"""
|
||||
Phase 0 — image_cache.py
|
||||
|
||||
SQLite-backed card image lookup service.
|
||||
|
||||
Provides:
|
||||
ImageCache — sync lookup (used by download scripts, CLI tools)
|
||||
AsyncImageCache — async lookup (used by FastAPI request handlers)
|
||||
|
||||
Lookup priority:
|
||||
1. SQLite index by scryfall_id (exact, fast)
|
||||
2. SQLite index by card name (case-insensitive, Forge card name format)
|
||||
3. CDN fallback URL (scryfall.io) — no local file, but still a valid URL
|
||||
|
||||
See issue #167
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
log = logging.getLogger("commander_ai_lab.cards.image_cache")
|
||||
|
||||
PROJECT_ROOT = Path(__file__).parent.parent.parent
|
||||
DB_PATH = PROJECT_ROOT / "data" / "card_image_index.db"
|
||||
STATIC_BASE_URL = "/static/card-images"
|
||||
|
||||
# Scryfall CDN URL pattern for fallback
|
||||
# Format: https://cards.scryfall.io/normal/front/{d1}/{d2}/{scryfall_id}.jpg
|
||||
# where d1/d2 are first two hex chars of the UUID
|
||||
def _cdn_url(scryfall_id: str, face: str = "front") -> str:
|
||||
"""Build a Scryfall CDN URL for a card face. Used as fallback."""
|
||||
d1 = scryfall_id[0] if scryfall_id else "0"
|
||||
d2 = scryfall_id[1] if len(scryfall_id) > 1 else "0"
|
||||
return f"https://cards.scryfall.io/normal/{face}/{d1}/{d2}/{scryfall_id}.jpg"
|
||||
|
||||
|
||||
class ImageCacheResult:
|
||||
"""Result of a card image lookup."""
|
||||
__slots__ = ("scryfall_id", "name_raw", "front_url", "back_url", "has_back", "is_local")
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
scryfall_id: str,
|
||||
name_raw: str,
|
||||
front_url: str,
|
||||
back_url: Optional[str],
|
||||
has_back: bool,
|
||||
is_local: bool,
|
||||
):
|
||||
self.scryfall_id = scryfall_id
|
||||
self.name_raw = name_raw
|
||||
self.front_url = front_url
|
||||
self.back_url = back_url
|
||||
self.has_back = has_back
|
||||
self.is_local = is_local
|
||||
|
||||
def to_dict(self) -> dict:
|
||||
return {
|
||||
"scryfall_id": self.scryfall_id,
|
||||
"name": self.name_raw,
|
||||
"art_url": self.front_url,
|
||||
"art_url_back": self.back_url,
|
||||
"has_back": self.has_back,
|
||||
"is_local": self.is_local,
|
||||
}
|
||||
|
||||
|
||||
class ImageCache:
|
||||
"""Synchronous SQLite-backed card image cache."""
|
||||
|
||||
def __init__(self, db_path: Path = DB_PATH, base_url: str = STATIC_BASE_URL):
|
||||
self._db_path = db_path
|
||||
self._base_url = base_url.rstrip("/")
|
||||
self._conn: Optional[sqlite3.Connection] = None
|
||||
|
||||
def _get_conn(self) -> sqlite3.Connection:
|
||||
if self._conn is None:
|
||||
if not self._db_path.exists():
|
||||
log.warning(
|
||||
"Image index DB not found at %s. Run scripts/build_image_index.py first.",
|
||||
self._db_path
|
||||
)
|
||||
return None
|
||||
self._conn = sqlite3.connect(str(self._db_path), check_same_thread=False)
|
||||
self._conn.row_factory = sqlite3.Row
|
||||
return self._conn
|
||||
|
||||
def _row_to_result(self, row: sqlite3.Row) -> ImageCacheResult:
|
||||
scryfall_id = row["scryfall_id"]
|
||||
front_path = row["front_path"]
|
||||
back_path = row["back_path"]
|
||||
has_back = bool(row["has_back"])
|
||||
|
||||
if front_path:
|
||||
front_url = f"{self._base_url}/{scryfall_id}_front.jpg"
|
||||
back_url = f"{self._base_url}/{scryfall_id}_back.jpg" if has_back else None
|
||||
is_local = True
|
||||
else:
|
||||
# CDN fallback
|
||||
front_url = _cdn_url(scryfall_id, "front")
|
||||
back_url = _cdn_url(scryfall_id, "back") if has_back else None
|
||||
is_local = False
|
||||
|
||||
return ImageCacheResult(
|
||||
scryfall_id=scryfall_id,
|
||||
name_raw=row["name_raw"],
|
||||
front_url=front_url,
|
||||
back_url=back_url,
|
||||
has_back=has_back,
|
||||
is_local=is_local,
|
||||
)
|
||||
|
||||
def lookup_by_id(self, scryfall_id: str) -> Optional[ImageCacheResult]:
|
||||
"""Look up card by Scryfall UUID."""
|
||||
conn = self._get_conn()
|
||||
if conn is None:
|
||||
return self._cdn_fallback(scryfall_id)
|
||||
row = conn.execute(
|
||||
"SELECT * FROM card_images WHERE scryfall_id = ?", (scryfall_id,)
|
||||
).fetchone()
|
||||
if row:
|
||||
return self._row_to_result(row)
|
||||
return self._cdn_fallback(scryfall_id)
|
||||
|
||||
def lookup_by_name(self, name: str) -> Optional[ImageCacheResult]:
|
||||
"""Look up card by name (case-insensitive). Returns most recent printing."""
|
||||
conn = self._get_conn()
|
||||
if conn is None:
|
||||
return None
|
||||
row = conn.execute(
|
||||
"SELECT * FROM card_images WHERE name = ? LIMIT 1",
|
||||
(name.lower().strip(),)
|
||||
).fetchone()
|
||||
if row:
|
||||
return self._row_to_result(row)
|
||||
return None
|
||||
|
||||
def _cdn_fallback(self, scryfall_id: str) -> ImageCacheResult:
|
||||
"""Return CDN URLs when local cache is unavailable."""
|
||||
return ImageCacheResult(
|
||||
scryfall_id=scryfall_id,
|
||||
name_raw="",
|
||||
front_url=_cdn_url(scryfall_id, "front"),
|
||||
back_url=None,
|
||||
has_back=False,
|
||||
is_local=False,
|
||||
)
|
||||
|
||||
def close(self):
|
||||
if self._conn:
|
||||
self._conn.close()
|
||||
self._conn = None
|
||||
|
||||
|
||||
# Module-level singleton for use by FastAPI route handlers
|
||||
_cache: Optional[ImageCache] = None
|
||||
|
||||
|
||||
def get_image_cache() -> ImageCache:
|
||||
"""Get the module-level ImageCache singleton."""
|
||||
global _cache
|
||||
if _cache is None:
|
||||
_cache = ImageCache()
|
||||
return _cache
|
||||
@@ -1 +0,0 @@
|
||||
# Phase 1: Forge subprocess bridge
|
||||
@@ -1,3 +0,0 @@
|
||||
# Phase 6 - game_result_logger.py
|
||||
# Captures winner, turn count, elimination order on Forge game end
|
||||
# See issue #173
|
||||
@@ -1,3 +0,0 @@
|
||||
# Phase 1 - game_state_parser.py
|
||||
# Converts Forge stdout text -> canonical GameStateDTO JSON
|
||||
# See issue #168
|
||||
@@ -1,4 +0,0 @@
|
||||
# Phase 1 - subprocess_manager.py
|
||||
# ForgeGameSession + ForgeSubprocessManager
|
||||
# Drives interactive Commander games via Forge stdin/stdout pipe
|
||||
# See issue #168
|
||||
@@ -1 +0,0 @@
|
||||
# Phase 6: ML Training Integration
|
||||
@@ -1,4 +0,0 @@
|
||||
# Phase 6 - training_data_builder.py
|
||||
# Converts decision_log + game_result -> (state, action, reward) tuples
|
||||
# Feeds existing PPO pipeline
|
||||
# See issue #173
|
||||
@@ -1,2 +0,0 @@
|
||||
# Runtime data directory
|
||||
# card_image_index.db, decision_logs/, game_results.jsonl, training_sets/
|
||||
+4
-20
@@ -43,8 +43,6 @@ from routes.coach import router as coach_router, init_coach_service
|
||||
from routes.ml import router as ml_router
|
||||
from routes.ws_game import router as ws_game_router
|
||||
from routes.game import router as game_router
|
||||
# Phase 0: card art endpoints
|
||||
from routes.card_art import router as card_art_router
|
||||
|
||||
log = logging.getLogger("commander_ai_lab.api")
|
||||
|
||||
@@ -53,7 +51,7 @@ _allowed = os.environ.get("ALLOWED_ORIGINS", _DEFAULT_ORIGINS)
|
||||
allowed_origins = [o.strip() for o in _allowed.split(",") if o.strip()]
|
||||
|
||||
|
||||
# ── Lifespan (replaces deprecated @app.on_event) ──────────────────────────────────────────────────
|
||||
# ── Lifespan (replaces deprecated @app.on_event) ────────────────────────────────────────────────────────
|
||||
@asynccontextmanager
|
||||
async def _lifespan(application: FastAPI):
|
||||
# ──── STARTUP ────
|
||||
@@ -138,8 +136,7 @@ app.include_router(deckgen_router)
|
||||
app.include_router(coach_router)
|
||||
app.include_router(ws_game_router)
|
||||
app.include_router(ml_router)
|
||||
app.include_router(game_router) # live game session routes
|
||||
app.include_router(card_art_router) # Phase 0: /card-art/* + /api/card-image-*
|
||||
app.include_router(game_router) # live game session routes
|
||||
|
||||
|
||||
# ── API endpoints ─────────────────────────────────────────────────────────────────────────────
|
||||
@@ -152,21 +149,8 @@ async def health_check():
|
||||
"""Health check endpoint for load balancers and Unity client polling."""
|
||||
return {"status": "ok"}
|
||||
|
||||
|
||||
# ── Static mounts (MUST come after all @app.get / router includes) ────────────────────────
|
||||
# ── Static UI (MUST come after all @app.get routes) ──────────────────────────────────────────
|
||||
from pathlib import Path as _Path
|
||||
|
||||
# Phase 0: Serve local card images at /static/card-images/
|
||||
# MUST be mounted before the legacy UI wildcard to avoid being shadowed.
|
||||
_card_images_dir = _Path(__file__).parent / "static" / "card-images"
|
||||
_card_images_dir.mkdir(parents=True, exist_ok=True)
|
||||
app.mount(
|
||||
"/static/card-images",
|
||||
StaticFiles(directory=str(_card_images_dir)),
|
||||
name="card-images"
|
||||
)
|
||||
|
||||
# Legacy UI / SPA (wildcard — MUST come last)
|
||||
_legacy_ui_dir = _Path(__file__).parent / "ui"
|
||||
_spa_dir = _Path(__file__).parent / "frontend" / "commander-ai-lab-ui" / "dist"
|
||||
|
||||
@@ -186,7 +170,7 @@ elif _spa_dir.exists():
|
||||
return FileResponse(str(_spa_dir / "index.html"))
|
||||
|
||||
|
||||
# ── Argument parsing + auto-detection helpers ────────────────────────────────────────────────
|
||||
# ── Argument parsing + auto-detection helpers ────────────────────────────────────────────────────────
|
||||
def _parse_args():
|
||||
p = argparse.ArgumentParser(description="Commander AI Lab API Server")
|
||||
p.add_argument("--forge-jar", default=os.environ.get("FORGE_JAR", ""))
|
||||
|
||||
@@ -9,7 +9,6 @@ httpx>=0.24.0
|
||||
websockets>=11.0
|
||||
openai>=1.0.0
|
||||
aiosqlite>=0.19.0
|
||||
aiofiles>=23.0.0
|
||||
|
||||
# Anthropic Claude API
|
||||
anthropic>=0.40.0
|
||||
@@ -20,9 +19,6 @@ beautifulsoup4>=4.12.0
|
||||
ollama>=0.2.0
|
||||
scrython>=2.0.0
|
||||
|
||||
# Phase 0: image download script
|
||||
tqdm>=4.65.0
|
||||
|
||||
# Embeddings
|
||||
|
||||
# RAG vector store
|
||||
|
||||
@@ -1,94 +0,0 @@
|
||||
"""
|
||||
Phase 0 — routes/card_art.py
|
||||
|
||||
FastAPI router for card art endpoints.
|
||||
|
||||
Endpoints:
|
||||
GET /card-art/{scryfall_id}
|
||||
Local-first image redirect.
|
||||
- If local file exists: redirect to /static/card-images/{id}_front.jpg
|
||||
- If not: redirect to Scryfall CDN (no 404, always returns an image)
|
||||
|
||||
GET /card-art/{scryfall_id}/back
|
||||
Same as above but for the back face of DFCs.
|
||||
|
||||
GET /api/card-image-info/{scryfall_id}
|
||||
JSON info: art_url, art_url_back, has_back, is_local
|
||||
|
||||
GET /api/card-image-by-name/{name}
|
||||
Look up by card name (case-insensitive). Returns image info JSON.
|
||||
|
||||
See issue #167
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import urllib.parse
|
||||
from typing import Optional
|
||||
|
||||
from fastapi import APIRouter, HTTPException
|
||||
from fastapi.responses import RedirectResponse
|
||||
|
||||
from commander_ai_lab.cards.image_cache import get_image_cache
|
||||
|
||||
router = APIRouter(tags=["card-art"])
|
||||
|
||||
|
||||
@router.get("/card-art/{scryfall_id}")
|
||||
async def get_card_art(scryfall_id: str):
|
||||
"""
|
||||
Redirect to the front face image for a card.
|
||||
Always returns a redirect — never a 404.
|
||||
Local cache first, Scryfall CDN fallback.
|
||||
"""
|
||||
cache = get_image_cache()
|
||||
result = cache.lookup_by_id(scryfall_id)
|
||||
if result:
|
||||
return RedirectResponse(url=result.front_url, status_code=302)
|
||||
# Ultimate fallback: redirect to CDN directly
|
||||
d1 = scryfall_id[0] if scryfall_id else "0"
|
||||
d2 = scryfall_id[1] if len(scryfall_id) > 1 else "0"
|
||||
cdn = f"https://cards.scryfall.io/normal/front/{d1}/{d2}/{scryfall_id}.jpg"
|
||||
return RedirectResponse(url=cdn, status_code=302)
|
||||
|
||||
|
||||
@router.get("/card-art/{scryfall_id}/back")
|
||||
async def get_card_art_back(scryfall_id: str):
|
||||
"""
|
||||
Redirect to the back face image for a DFC.
|
||||
Falls back to CDN back URL if not locally cached.
|
||||
"""
|
||||
cache = get_image_cache()
|
||||
result = cache.lookup_by_id(scryfall_id)
|
||||
if result and result.back_url:
|
||||
return RedirectResponse(url=result.back_url, status_code=302)
|
||||
d1 = scryfall_id[0] if scryfall_id else "0"
|
||||
d2 = scryfall_id[1] if len(scryfall_id) > 1 else "0"
|
||||
cdn = f"https://cards.scryfall.io/normal/back/{d1}/{d2}/{scryfall_id}.jpg"
|
||||
return RedirectResponse(url=cdn, status_code=302)
|
||||
|
||||
|
||||
@router.get("/api/card-image-info/{scryfall_id}")
|
||||
async def get_card_image_info(scryfall_id: str):
|
||||
"""
|
||||
Return JSON image info for a card by Scryfall ID.
|
||||
Always returns a result (CDN fallback if not cached locally).
|
||||
"""
|
||||
cache = get_image_cache()
|
||||
result = cache.lookup_by_id(scryfall_id)
|
||||
if result:
|
||||
return result.to_dict()
|
||||
raise HTTPException(status_code=404, detail=f"Card not found: {scryfall_id}")
|
||||
|
||||
|
||||
@router.get("/api/card-image-by-name/{name}")
|
||||
async def get_card_image_by_name(name: str):
|
||||
"""
|
||||
Return JSON image info for a card by name (case-insensitive).
|
||||
Uses the SQLite index. Returns 404 if name not found.
|
||||
"""
|
||||
decoded_name = urllib.parse.unquote(name)
|
||||
cache = get_image_cache()
|
||||
result = cache.lookup_by_name(decoded_name)
|
||||
if result:
|
||||
return result.to_dict()
|
||||
raise HTTPException(status_code=404, detail=f"Card name not found: {decoded_name}")
|
||||
+35
-2
@@ -57,6 +57,32 @@ def _sanitize_floats(obj):
|
||||
return obj
|
||||
|
||||
|
||||
def _inject_win_type(data: dict, win_type_override: str) -> None:
|
||||
"""Stamp winType on every per-game dict that doesn't already have one.
|
||||
|
||||
Called by get_result() after loading the JSON from disk.
|
||||
- Uses setdefault so DeepSeek per-game values (already set in forge_runner.py)
|
||||
are never overwritten.
|
||||
- Also stamps metadata.winType for a quick batch-level summary field.
|
||||
- win_type_override comes from state.win_type_override (Java path) or
|
||||
defaults to "Combat" if the attribute is absent (DeepSeek / old batches).
|
||||
"""
|
||||
if not win_type_override or win_type_override == "Combat":
|
||||
# Still stamp Combat as default so the field is always present
|
||||
win_type_override = "Combat"
|
||||
|
||||
# Batch-level metadata field
|
||||
meta = data.get("metadata")
|
||||
if isinstance(meta, dict):
|
||||
meta.setdefault("winType", win_type_override)
|
||||
|
||||
# Per-game dicts nested under data["decks"][N]["games"][M]
|
||||
for deck in data.get("decks", []):
|
||||
for game in deck.get("games", []):
|
||||
if isinstance(game, dict):
|
||||
game.setdefault("winType", win_type_override)
|
||||
|
||||
|
||||
@router.post("/api/lab/start", response_model=StartResponse)
|
||||
async def start_batch(req: StartRequest, background_tasks: BackgroundTasks):
|
||||
"""Start a new batch simulation run."""
|
||||
@@ -196,6 +222,15 @@ async def get_result(batchId: Optional[str] = None):
|
||||
data = json.load(f)
|
||||
except (json.JSONDecodeError, OSError) as e:
|
||||
raise HTTPException(500, f"Failed to read result file: {e}")
|
||||
|
||||
# ── Inject winType into every game dict (fix #180) ──────────────────────
|
||||
# win_type_override is set by forge_runner._run_process_blocking() for the
|
||||
# Java/Forge path. DeepSeek games already have per-game winType stamped
|
||||
# directly in the result dict; _inject_win_type uses setdefault so those
|
||||
# are never overwritten.
|
||||
win_type_override = getattr(state, "win_type_override", "Combat")
|
||||
_inject_win_type(data, win_type_override)
|
||||
|
||||
data = _sanitize_floats(data)
|
||||
return JSONResponse(content=data)
|
||||
|
||||
@@ -469,5 +504,3 @@ async def start_batch_deepseek(request: FastAPIRequest, background_tasks: Backgr
|
||||
'message': f'Running {num_games} games across {len(decks)} decks with DeepSeek AI',
|
||||
'engine': 'deepseek',
|
||||
})
|
||||
|
||||
|
||||
@@ -1,135 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Phase 0 — build_image_index.py
|
||||
|
||||
Builds (or rebuilds) the SQLite card image index at:
|
||||
data/card_image_index.db
|
||||
|
||||
Schema:
|
||||
card_images(
|
||||
scryfall_id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL, -- card name, lowercased for lookup
|
||||
name_raw TEXT NOT NULL, -- original casing
|
||||
front_path TEXT, -- relative path from project root
|
||||
back_path TEXT, -- DFC back face, NULL for single-faced
|
||||
has_back INTEGER DEFAULT 0 -- 1 if DFC
|
||||
)
|
||||
CREATE INDEX idx_name ON card_images(name);
|
||||
|
||||
Usage:
|
||||
python scripts/build_image_index.py
|
||||
python scripts/build_image_index.py --rebuild
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import sqlite3
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
log = logging.getLogger("build_image_index")
|
||||
|
||||
SCRYFALL_BULK_API = "https://api.scryfall.com/bulk-data/default-cards"
|
||||
PROJECT_ROOT = Path(__file__).parent.parent
|
||||
IMAGE_DIR = PROJECT_ROOT / "static" / "card-images"
|
||||
DB_PATH = PROJECT_ROOT / "data" / "card_image_index.db"
|
||||
|
||||
CREATE_TABLE_SQL = """
|
||||
CREATE TABLE IF NOT EXISTS card_images (
|
||||
scryfall_id TEXT PRIMARY KEY,
|
||||
name TEXT NOT NULL,
|
||||
name_raw TEXT NOT NULL,
|
||||
front_path TEXT,
|
||||
back_path TEXT,
|
||||
has_back INTEGER DEFAULT 0
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_name ON card_images(name);
|
||||
"""
|
||||
|
||||
|
||||
def fetch_bulk_cards() -> list[dict]:
|
||||
log.info("Fetching Scryfall bulk manifest...")
|
||||
with urllib.request.urlopen(SCRYFALL_BULK_API, timeout=30) as r:
|
||||
meta = json.loads(r.read())
|
||||
url = meta["download_uri"]
|
||||
log.info("Downloading bulk card data...")
|
||||
with urllib.request.urlopen(url, timeout=120) as r:
|
||||
return json.loads(r.read())
|
||||
|
||||
|
||||
def build_rows(cards: list[dict]) -> list[tuple]:
|
||||
"""
|
||||
For each card, check if the image file(s) exist locally.
|
||||
Builds rows for both single-faced and DFC cards.
|
||||
Tokens are included if they have image files.
|
||||
"""
|
||||
rows: list[tuple] = []
|
||||
|
||||
for card in cards:
|
||||
scryfall_id = card.get("id", "")
|
||||
name_raw = card.get("name", "")
|
||||
if not scryfall_id or not name_raw:
|
||||
continue
|
||||
|
||||
name = name_raw.lower().strip()
|
||||
|
||||
front_file = IMAGE_DIR / f"{scryfall_id}_front.jpg"
|
||||
back_file = IMAGE_DIR / f"{scryfall_id}_back.jpg"
|
||||
|
||||
front_path = f"static/card-images/{scryfall_id}_front.jpg" if front_file.exists() else None
|
||||
back_path = f"static/card-images/{scryfall_id}_back.jpg" if back_file.exists() else None
|
||||
has_back = 1 if back_path else 0
|
||||
|
||||
rows.append((scryfall_id, name, name_raw, front_path, back_path, has_back))
|
||||
|
||||
return rows
|
||||
|
||||
|
||||
def write_db(rows: list[tuple], rebuild: bool) -> None:
|
||||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
if rebuild and DB_PATH.exists():
|
||||
DB_PATH.unlink()
|
||||
log.info("Deleted existing database for rebuild.")
|
||||
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
try:
|
||||
conn.executescript(CREATE_TABLE_SQL)
|
||||
conn.executemany(
|
||||
"""
|
||||
INSERT OR REPLACE INTO card_images
|
||||
(scryfall_id, name, name_raw, front_path, back_path, has_back)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
""",
|
||||
rows
|
||||
)
|
||||
conn.commit()
|
||||
log.info("Indexed %d cards into %s", len(rows), DB_PATH)
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Build SQLite card image index.")
|
||||
parser.add_argument("--rebuild", action="store_true",
|
||||
help="Drop and rebuild the database from scratch")
|
||||
parser.add_argument("--verbose", "-v", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s"
|
||||
)
|
||||
|
||||
cards = fetch_bulk_cards()
|
||||
log.info("Processing %d card entries...", len(cards))
|
||||
rows = build_rows(cards)
|
||||
log.info("Found %d cards with local images.", len(rows))
|
||||
write_db(rows, args.rebuild)
|
||||
log.info("Index build complete: %s", DB_PATH)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,181 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Phase 0 — download_card_images.py
|
||||
|
||||
Bulk-downloads all MTG card images from Scryfall's default_cards bulk export.
|
||||
Stores images at: static/card-images/{scryfall_id}_front.jpg
|
||||
static/card-images/{scryfall_id}_back.jpg (DFCs only)
|
||||
|
||||
Usage:
|
||||
python scripts/download_card_images.py
|
||||
python scripts/download_card_images.py --image-size normal
|
||||
python scripts/download_card_images.py --workers 8 --resume
|
||||
|
||||
Scryfall bulk image domain has no rate limits, so no sleep is needed.
|
||||
See: https://scryfall.com/docs/api/bulk-data
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
from tqdm import tqdm
|
||||
except ImportError:
|
||||
tqdm = None
|
||||
|
||||
log = logging.getLogger("download_card_images")
|
||||
|
||||
SCRYFALL_BULK_API = "https://api.scryfall.com/bulk-data/default-cards"
|
||||
DEFAULT_IMAGE_SIZE = "normal" # 488x680px — ideal for Unity card rendering
|
||||
DEFAULT_WORKERS = 6
|
||||
|
||||
# Scryfall requires a descriptive User-Agent — returns HTTP 400 without it.
|
||||
# See: https://scryfall.com/docs/api#making-requests
|
||||
USER_AGENT = "CommanderAILab/1.0 (https://github.com/KoalaTrapLord/commander-ai-lab)"
|
||||
|
||||
PROJECT_ROOT = Path(__file__).parent.parent
|
||||
OUTPUT_DIR = PROJECT_ROOT / "static" / "card-images"
|
||||
|
||||
|
||||
def _open(url: str, timeout: int = 30):
|
||||
"""urllib.request.urlopen with required Scryfall User-Agent header."""
|
||||
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||||
return urllib.request.urlopen(req, timeout=timeout)
|
||||
|
||||
|
||||
def fetch_bulk_download_url(image_size: str) -> str:
|
||||
"""Get the Scryfall bulk data download URL for default_cards."""
|
||||
log.info("Fetching Scryfall bulk data manifest...")
|
||||
with _open(SCRYFALL_BULK_API, timeout=30) as resp:
|
||||
meta = json.loads(resp.read())
|
||||
return meta["download_uri"]
|
||||
|
||||
|
||||
def load_bulk_data(download_url: str) -> list[dict]:
|
||||
"""Stream-download and parse the bulk JSON (can be ~100MB)."""
|
||||
log.info("Downloading bulk card data from %s ...", download_url)
|
||||
with _open(download_url, timeout=120) as resp:
|
||||
cards = json.loads(resp.read())
|
||||
log.info("Loaded %d card entries.", len(cards))
|
||||
return cards
|
||||
|
||||
|
||||
def get_image_jobs(cards: list[dict], image_size: str, resume: bool) -> list[tuple[str, str]]:
|
||||
"""
|
||||
Build a list of (url, dest_path) tuples to download.
|
||||
Handles:
|
||||
- Single-faced cards: {scryfall_id}_front.jpg
|
||||
- Double-faced cards (DFCs): {scryfall_id}_front.jpg + {scryfall_id}_back.jpg
|
||||
- Tokens: uses image_uris if present, else skips gracefully
|
||||
"""
|
||||
jobs: list[tuple[str, str]] = []
|
||||
|
||||
for card in cards:
|
||||
scryfall_id = card.get("id", "")
|
||||
if not scryfall_id:
|
||||
continue
|
||||
|
||||
# Single-faced or meld cards
|
||||
if "image_uris" in card:
|
||||
url = card["image_uris"].get(image_size)
|
||||
if url:
|
||||
dest = str(OUTPUT_DIR / f"{scryfall_id}_front.jpg")
|
||||
if not resume or not Path(dest).exists():
|
||||
jobs.append((url, dest))
|
||||
|
||||
# Double-faced cards (DFCs): modal, transform, adventure, etc.
|
||||
elif "card_faces" in card:
|
||||
for idx, face in enumerate(card["card_faces"]):
|
||||
face_uris = face.get("image_uris", {})
|
||||
url = face_uris.get(image_size)
|
||||
if url:
|
||||
side = "front" if idx == 0 else "back"
|
||||
dest = str(OUTPUT_DIR / f"{scryfall_id}_{side}.jpg")
|
||||
if not resume or not Path(dest).exists():
|
||||
jobs.append((url, dest))
|
||||
|
||||
return jobs
|
||||
|
||||
|
||||
def download_image(url: str, dest: str) -> tuple[bool, str]:
|
||||
"""Download a single image. Returns (success, dest)."""
|
||||
try:
|
||||
req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
|
||||
with urllib.request.urlopen(req, timeout=30) as resp:
|
||||
with open(dest, "wb") as f:
|
||||
f.write(resp.read())
|
||||
return True, dest
|
||||
except Exception as exc:
|
||||
log.warning("Failed to download %s -> %s: %s", url, dest, exc)
|
||||
return False, dest
|
||||
|
||||
|
||||
def run_downloads(jobs: list[tuple[str, str]], workers: int) -> tuple[int, int]:
|
||||
"""Download all jobs concurrently. Returns (success_count, fail_count)."""
|
||||
success = 0
|
||||
fail = 0
|
||||
|
||||
iterator = tqdm(total=len(jobs), unit="img", desc="Downloading") if tqdm else None
|
||||
|
||||
with ThreadPoolExecutor(max_workers=workers) as pool:
|
||||
futures = {pool.submit(download_image, url, dest): dest for url, dest in jobs}
|
||||
for future in as_completed(futures):
|
||||
ok, _ = future.result()
|
||||
if ok:
|
||||
success += 1
|
||||
else:
|
||||
fail += 1
|
||||
if iterator:
|
||||
iterator.update(1)
|
||||
|
||||
if iterator:
|
||||
iterator.close()
|
||||
|
||||
return success, fail
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Bulk download Scryfall card images.")
|
||||
parser.add_argument("--image-size", default=DEFAULT_IMAGE_SIZE,
|
||||
choices=["small", "normal", "large", "png", "art_crop", "border_crop"],
|
||||
help="Image size to download (default: normal = 488x680px)")
|
||||
parser.add_argument("--workers", type=int, default=DEFAULT_WORKERS,
|
||||
help="Concurrent download threads (default: 6)")
|
||||
parser.add_argument("--resume", action="store_true",
|
||||
help="Skip images that already exist on disk")
|
||||
parser.add_argument("--verbose", "-v", action="store_true")
|
||||
args = parser.parse_args()
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.DEBUG if args.verbose else logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(message)s"
|
||||
)
|
||||
|
||||
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
|
||||
log.info("Output directory: %s", OUTPUT_DIR)
|
||||
|
||||
download_url = fetch_bulk_download_url(args.image_size)
|
||||
cards = load_bulk_data(download_url)
|
||||
jobs = get_image_jobs(cards, args.image_size, args.resume)
|
||||
|
||||
log.info("Jobs to download: %d (resume=%s)", len(jobs), args.resume)
|
||||
if not jobs:
|
||||
log.info("Nothing to do. All images already present.")
|
||||
return
|
||||
|
||||
success, fail = run_downloads(jobs, args.workers)
|
||||
log.info("Done. Success: %d Failed: %d Total: %d", success, fail, len(jobs))
|
||||
if fail > 0:
|
||||
log.warning("%d images failed — re-run with --resume to retry.", fail)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,4 +0,0 @@
|
||||
# Phase 6 - run_self_play.py
|
||||
# Headless self-play runner: all 4 players AI-driven, no Unity needed
|
||||
# Usage: python scripts/run_self_play.py --num-games 500 --deck-pool-path data/decks/
|
||||
# See issue #173
|
||||
@@ -262,6 +262,23 @@ def _run_process_blocking(state: BatchState, cmd: list):
|
||||
state.running = False
|
||||
state.completed_games = state.total_games
|
||||
|
||||
# ── Post-process: detect alternate win conditions in the Java log ──
|
||||
# Runs after the process exits so we don't slow the hot stdout loop.
|
||||
# Stamps state.win_type_override which routes/lab_api.py can attach
|
||||
# to each game result when building the batch JSON response.
|
||||
try:
|
||||
src_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'src')
|
||||
src_dir = os.path.normpath(src_dir)
|
||||
if src_dir not in sys.path:
|
||||
sys.path.insert(0, src_dir)
|
||||
from commander_ai_lab.sim.win_condition_parser import detect_win_type_from_log
|
||||
state.win_type_override = detect_win_type_from_log(state.log_lines)
|
||||
if state.win_type_override != "Combat":
|
||||
log.info(f"[WinType] Alternate win detected in batch {state.batch_id}: {state.win_type_override}")
|
||||
except Exception as _wcp_err:
|
||||
log.debug(f"[WinType] Parser import failed (non-fatal): {_wcp_err}")
|
||||
state.win_type_override = "Combat"
|
||||
|
||||
|
||||
async def run_batch_subprocess(
|
||||
state: BatchState,
|
||||
@@ -344,6 +361,7 @@ def _run_deepseek_batch_thread(
|
||||
from commander_ai_lab.sim.models import Card
|
||||
from commander_ai_lab.sim.deepseek_engine import DeepSeekGameEngine
|
||||
from commander_ai_lab.sim.rules import enrich_card
|
||||
from commander_ai_lab.sim.win_condition_parser import detect_win_type_from_cards
|
||||
|
||||
state.log_lines.append('[DeepSeek Batch] Initializing DeepSeek brain...')
|
||||
brain = _get_deepseek_brain()
|
||||
@@ -503,6 +521,31 @@ def _run_deepseek_batch_thread(
|
||||
)
|
||||
gd = result.to_dict()
|
||||
gd['gameNumber'] = g + 1
|
||||
|
||||
# ── Alternate win detection (DeepSeek path) ──────────────
|
||||
# Inspect the winning player's battlefield for alternate-win
|
||||
# permanents. result.winner is the seat index (0 = deck_a).
|
||||
try:
|
||||
winner_seat = result.winner_seat
|
||||
if winner_seat >= 0:
|
||||
from commander_ai_lab.sim.game_state import get_final_battlefield
|
||||
winner_bf = getattr(result, '_winner_battlefield', None)
|
||||
if winner_bf is not None:
|
||||
win_type = detect_win_type_from_cards(winner_bf)
|
||||
else:
|
||||
# Fallback: scan game_log lines if battlefield snapshot unavailable
|
||||
from commander_ai_lab.sim.win_condition_parser import detect_win_type_from_log
|
||||
win_type = detect_win_type_from_log(
|
||||
[str(entry) for entry in (result.game_log or [])]
|
||||
)
|
||||
result.win_type = win_type
|
||||
gd['winType'] = win_type
|
||||
if win_type != 'Combat':
|
||||
log.info(f'[WinType] {deck_name} game {g+1}: {win_type} win detected')
|
||||
except Exception as _wt_err:
|
||||
log.debug(f'[WinType] Detection failed for game {g+1} (non-fatal): {_wt_err}')
|
||||
gd.setdefault('winType', 'Combat')
|
||||
|
||||
deck_stats['games'].append(gd)
|
||||
|
||||
if result.winner == 0:
|
||||
|
||||
@@ -354,11 +354,16 @@ class GameResult:
|
||||
|
||||
- players: list of PlayerResult, one per seat
|
||||
- winner_seat: index of the winning player (-1 = draw)
|
||||
- win_type: how the game was won — "Combat", "Combo", "Drain", "Mill",
|
||||
"Poison", or "CmdrDmg". Defaults to "Combat".
|
||||
Set by win_condition_parser.detect_win_type_from_cards() (Python engine)
|
||||
or win_condition_parser.detect_win_type_from_log() (Java/Forge engine).
|
||||
- Backward-compat properties player_a_stats / player_b_stats delegate to players[0] / players[1]
|
||||
"""
|
||||
|
||||
winner_seat: int = -1
|
||||
turns: int = 0
|
||||
win_type: str = "Combat" # populated by win_condition_parser
|
||||
players: list[PlayerResult] = field(default_factory=list)
|
||||
game_log: list = field(default_factory=list)
|
||||
|
||||
@@ -428,6 +433,7 @@ class GameResult:
|
||||
# Legacy keys (UI expects these)
|
||||
"winner": self.winner_seat,
|
||||
"turns": self.turns,
|
||||
"winType": self.win_type,
|
||||
# New N-player keys (Java BatchResult schema)
|
||||
"winningSeat": self.winner_seat,
|
||||
"playerResults": [p.to_dict() for p in self.players],
|
||||
|
||||
@@ -0,0 +1,155 @@
|
||||
"""
|
||||
Commander AI Lab — Alternate Win Condition Parser
|
||||
==================================================
|
||||
Scans Forge log lines or oracle text for non-combat "you win the game"
|
||||
triggers and returns a win type category string.
|
||||
|
||||
Used by:
|
||||
- services/forge_runner.py (Java batch path: post-process log_lines)
|
||||
- src/commander_ai_lab/sim/deepseek_engine.py (DeepSeek path: check oracle text)
|
||||
- src/commander_ai_lab/sim/engine.py (Python sim path: check oracle text)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Alternate win card registry
|
||||
# ---------------------------------------------------------------------------
|
||||
# Maps lowercase card name -> win category.
|
||||
# Categories: "Combo", "Drain", "Mill", "Poison"
|
||||
# "Combat" is the default and should never appear here.
|
||||
|
||||
ALTERNATE_WIN_CARDS: dict[str, str] = {
|
||||
# Spell-based
|
||||
"approach of the second sun": "Combo",
|
||||
"thassa's oracle": "Combo",
|
||||
"jace, wielder of mysteries": "Combo",
|
||||
"laboratory maniac": "Combo",
|
||||
"leveler": "Combo", # pairs with Lab Man / Thassa
|
||||
"gleemax": "Combo",
|
||||
"coalition victory": "Combo",
|
||||
"helix pinnacle": "Combo",
|
||||
"mayael's aria": "Combo",
|
||||
"mechanized production": "Combo",
|
||||
"happily ever after": "Combo",
|
||||
"simic ascendancy": "Combo",
|
||||
"azor's elocutors": "Combo",
|
||||
"chance encounter": "Combo",
|
||||
"darksteel reactor": "Combo",
|
||||
"hellkite tyrant": "Combo",
|
||||
"mortal kombat": "Combo",
|
||||
"revel in riches": "Drain",
|
||||
"test of endurance": "Combo",
|
||||
"felidar sovereign": "Combo",
|
||||
"celestial convergence": "Combo",
|
||||
"barren glory": "Combo",
|
||||
"maze's end": "Combo",
|
||||
"new horizons": "Combo",
|
||||
"thespian's stage": "Combo", # copying Maze's End
|
||||
# Poison / Infect
|
||||
"phyrexian unlife": "Poison",
|
||||
"poisoned tips": "Poison",
|
||||
# Mill
|
||||
"milling": "Mill", # generic tag
|
||||
"the fall of lord karsus": "Mill",
|
||||
}
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Forge log scanner
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Matches lines like:
|
||||
# [COMBAT-LINE] Resolve Stack: Approach of the Second Sun ... you win the game.
|
||||
# [OUTCOME-LINE] ... you win the game ...
|
||||
# Game Result: Player won (Approach of the Second Sun)
|
||||
_YOU_WIN_RE = re.compile(
|
||||
r'you win the game',
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
# Extracts a card name from the line so we can look it up in ALTERNATE_WIN_CARDS.
|
||||
# Forge typically formats these as "Resolve Stack: <CardName> ..." or
|
||||
# "Player won via <CardName>".
|
||||
_CARD_NAME_RE = re.compile(
|
||||
r'(?:Resolve Stack|via|won by|triggered by|winning card)[:\s]+([A-Za-z\'\,\.\- ]+)',
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def detect_win_type_from_log(log_lines: list[str]) -> str:
|
||||
"""
|
||||
Scan a list of Forge stdout log lines for alternate win condition triggers.
|
||||
|
||||
Returns the win category ("Combo", "Drain", "Mill", "Poison") if an
|
||||
alternate win card is detected, otherwise returns "Combat".
|
||||
|
||||
Scans *all* lines so the last detected alternate win takes precedence
|
||||
(handles edge cases where the log mentions multiple near-wins).
|
||||
"""
|
||||
result = "Combat"
|
||||
for line in log_lines:
|
||||
if not _YOU_WIN_RE.search(line):
|
||||
continue
|
||||
# Try to extract the card name responsible
|
||||
m = _CARD_NAME_RE.search(line)
|
||||
if m:
|
||||
card_name = m.group(1).strip().rstrip('.').lower()
|
||||
# Exact lookup first
|
||||
if card_name in ALTERNATE_WIN_CARDS:
|
||||
result = ALTERNATE_WIN_CARDS[card_name]
|
||||
continue
|
||||
# Partial / substring match for long card names
|
||||
for known, category in ALTERNATE_WIN_CARDS.items():
|
||||
if known in card_name or card_name in known:
|
||||
result = category
|
||||
break
|
||||
else:
|
||||
# No card name found but "you win the game" is present —
|
||||
# scan the full line for any known card name
|
||||
line_lower = line.lower()
|
||||
for known, category in ALTERNATE_WIN_CARDS.items():
|
||||
if known in line_lower:
|
||||
result = category
|
||||
break
|
||||
return result
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Oracle text scanner (Python / DeepSeek engine path)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def detect_win_type_from_cards(cards_on_battlefield: list) -> str:
|
||||
"""
|
||||
Inspect a player's battlefield (list of Card objects) for alternate win
|
||||
permanents. Returns the win category or "Combat".
|
||||
|
||||
Checks:
|
||||
1. Card name against ALTERNATE_WIN_CARDS (fastest)
|
||||
2. oracle_text for "you win the game" as a safety net
|
||||
"""
|
||||
for card in cards_on_battlefield:
|
||||
name_lower = getattr(card, 'name', '').lower()
|
||||
if name_lower in ALTERNATE_WIN_CARDS:
|
||||
return ALTERNATE_WIN_CARDS[name_lower]
|
||||
oracle = getattr(card, 'oracle_text', '') or ''
|
||||
if 'you win the game' in oracle.lower():
|
||||
# Unknown alternate win card — still not Combat
|
||||
return "Combo"
|
||||
return "Combat"
|
||||
|
||||
|
||||
def detect_win_type_from_oracle(oracle_text: str, card_name: str = "") -> str:
|
||||
"""
|
||||
Simple oracle text probe used when we have a resolved spell/ability
|
||||
but not a full battlefield list.
|
||||
"""
|
||||
if card_name:
|
||||
lookup = card_name.strip().lower()
|
||||
if lookup in ALTERNATE_WIN_CARDS:
|
||||
return ALTERNATE_WIN_CARDS[lookup]
|
||||
if 'you win the game' in (oracle_text or '').lower():
|
||||
return "Combo"
|
||||
return "Combat"
|
||||
@@ -1,2 +0,0 @@
|
||||
# Phase 0: Card images downloaded here by scripts/download_card_images.py
|
||||
# ~1.5-2GB at normal size
|
||||
@@ -1,2 +0,0 @@
|
||||
# Phase 4: Unity TCGEngine project lives here
|
||||
# See issue #171
|
||||
Reference in new issue
Block a user