diff --git "a/app.py" "b/app.py"
--- "a/app.py"
+++ "b/app.py"
@@ -1,1599 +1,1645 @@
-"""Hugging Face Space wrapper for Vawlrathh.
-
-This module provides a pure Gradio application for deployment on Hugging Face Spaces
-with ZeroGPU support. It integrates directly with Vawlrathh services instead of
-using a separate FastAPI backend, ensuring compatibility with the Gradio SDK runner.
-
-"Your deck's terrible. Let me show you how to fix it."
-— Vawlrathh, The Small'n
-"""
-
-import logging
-import math
-import os
-import sys
-import textwrap
-import time
-import json
-import asyncio
-from dataclasses import dataclass
-from typing import Any, Callable, Dict, List, Optional, Tuple
-
-from dotenv import load_dotenv
-import gradio as gr
-import spaces
-
-# Load environment variables from .env file
-load_dotenv()
-
-# Ensure project root is in python path
-sys.path.append(os.path.dirname(os.path.abspath(__file__)))
-
-# Import Services
-from src.services.smart_sql import SmartSQLService
-from src.services.deck_analyzer import DeckAnalyzer
-from src.services.meta_intelligence import MetaIntelligenceService
-from src.services.chat_agent import ConcurrentChatService
-from src.services.smart_inference import SmartInferenceService
-from src.services.card_market_service import CardMarketService
-from src.utils.csv_parser import parse_arena_csv, parse_deck_string
-
-# Configure logging
-logging.basicConfig(
- level=logging.INFO,
- format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-)
-logger = logging.getLogger(__name__)
-
-# -----------------------------------------------------------------------------
-# Service Initialization
-# -----------------------------------------------------------------------------
-
-# Initialize services
-# Note: Async initialization happens in the event loop
-sql_service = SmartSQLService()
-meta_service = MetaIntelligenceService()
-deck_analyzer = DeckAnalyzer(meta_service=meta_service)
-chat_service = ConcurrentChatService()
-inference_service = SmartInferenceService()
-card_market_service = CardMarketService()
-
-# -----------------------------------------------------------------------------
-# GPU / Spaces Configuration
-# -----------------------------------------------------------------------------
-
-@spaces.GPU(duration=10)
-def initialize_gpu():
- """Initialize GPU runtime for HF Spaces ZERO.
-
- This function exists primarily to satisfy the ZeroGPU requirement that
- at least one function must be decorated with @spaces.GPU.
- """
- import torch
- if torch.cuda.is_available():
- device = torch.cuda.get_device_name(0)
- logger.info(f"GPU initialized: {device}")
- return {"gpu": device, "cuda_available": True}
- return {"gpu": None, "cuda_available": False}
-
-# -----------------------------------------------------------------------------
-# Helper Functions (Direct Service Calls)
-# -----------------------------------------------------------------------------
-
-async def ensure_db_initialized():
- """Ensure database is initialized."""
- await sql_service.init_db()
-
-async def handle_csv_upload(uploaded_file, previous_id):
- """Handle CSV upload by calling services directly."""
- if not uploaded_file:
- return {"status": "error", "message": "No CSV file selected"}, previous_id, previous_id
-
- try:
- # Ensure DB is ready
- await ensure_db_initialized()
-
- # Handle both file path (Gradio 4.x+) and file object (older Gradio)
- if hasattr(uploaded_file, 'read'):
- content = uploaded_file.read()
- else:
- # uploaded_file is a path string (NamedString or str)
- with open(str(uploaded_file), 'r', encoding='utf-8') as f:
- content = f.read()
- if isinstance(content, bytes):
- content = content.decode('utf-8')
-
- deck = parse_arena_csv(content)
- deck_id = await sql_service.store_deck(deck)
-
- return {
- "status": "success",
- "message": f"Deck '{deck.name}' uploaded successfully",
- "deck_id": deck_id
- }, deck_id, deck_id
- except Exception as e:
- logger.exception("CSV upload failed")
- return {"status": "error", "message": str(e)}, previous_id, previous_id
-
-async def handle_text_upload(deck_text, fmt, previous_id):
- """Handle text upload by calling services directly."""
- if not deck_text or not deck_text.strip():
- return {"status": "error", "message": "Deck text is empty"}, previous_id, previous_id
-
- try:
- # Ensure DB is ready
- await ensure_db_initialized()
-
- deck = parse_deck_string(deck_text)
- deck.format = fmt
- deck_id = await sql_service.store_deck(deck)
-
- return {
- "status": "success",
- "message": f"Deck uploaded successfully",
- "deck_id": deck_id
- }, deck_id, deck_id
- except Exception as e:
- logger.exception("Text upload failed")
- return {"status": "error", "message": str(e)}, previous_id, previous_id
-
-async def handle_analyze(deck_id: float):
- """Analyze a deck using DeckAnalyzer."""
- if not deck_id:
- return {"status": "error", "message": "No deck ID provided"}
-
- try:
- deck = await sql_service.get_deck(int(deck_id))
- if not deck:
- return {"status": "error", "message": "Deck not found"}
-
- analysis = await deck_analyzer.analyze_deck(deck)
-
- # Convert to dict for JSON output
- return {
- "deck_name": analysis.deck_name,
- "overall_score": analysis.overall_score,
- "mana_curve": analysis.mana_curve.__dict__,
- "strengths": analysis.strengths,
- "weaknesses": analysis.weaknesses,
- "synergies": [s.__dict__ for s in analysis.synergies],
- "meta_matchups": [m.__dict__ for m in analysis.meta_matchups]
- }
- except Exception as e:
- logger.exception("Analysis failed")
- return {"status": "error", "message": str(e)}
-
-async def handle_optimize(deck_id: float):
- """Generate optimization suggestions."""
- if not deck_id:
- return {"status": "error", "message": "No deck ID provided"}
-
- try:
- deck = await sql_service.get_deck(int(deck_id))
- if not deck:
- return {"status": "error", "message": "Deck not found"}
-
- analysis = await deck_analyzer.analyze_deck(deck)
- suggestions = await inference_service.generate_suggestions(deck, analysis)
-
- return [s.__dict__ for s in suggestions]
- except Exception as e:
- logger.exception("Optimization failed")
- return {"status": "error", "message": str(e)}
-
-async def handle_meta_snapshot(game_format: str):
- """Fetch meta snapshot."""
- try:
- snapshot = await meta_service.get_current_meta(game_format)
- return meta_service.to_dict(snapshot)
- except Exception as e:
- logger.exception("Meta snapshot failed")
- return {"status": "error", "message": str(e)}
-
-async def handle_memory_summary(deck_id: float):
- """Fetch deck performance history."""
- if not deck_id:
- return {"status": "error", "message": "No deck ID provided"}
-
- try:
- history = await sql_service.get_deck_performance(int(deck_id))
- return history
- except Exception as e:
- logger.exception("Memory summary failed")
- return {"status": "error", "message": str(e)}
-
-async def handle_collection_upload(
- uploaded_file,
- previous_id,
- progress=gr.Progress(track_tqdm=True)
-):
- """
- Handle large collection CSV upload with chunked processing and progress tracking.
-
- Args:
- uploaded_file: File path from gr.File component
- previous_id: Previous collection ID from state
- progress: Gradio progress tracker
-
- Returns:
- Tuple of (status_dict, collection_id, collection_id)
- """
- from src.utils.csv_parser import count_csv_rows, parse_arena_csv_chunked
- from src.models.deck import Collection, CollectionProcessingResult, Card
-
- start_time = time.time()
-
- # Validate input
- if uploaded_file is None:
- return {
- "status": "error",
- "message": "No file uploaded. Please select a CSV file."
- }, previous_id, previous_id
-
- # Get file path (Gradio 4.x+ returns path string)
- filepath = uploaded_file if isinstance(uploaded_file, str) else uploaded_file.name
-
- try:
- # Phase 1: Count rows for progress calculation
- progress(0, desc="Analyzing file size...")
- total_rows = count_csv_rows(filepath)
-
- if total_rows == 0:
- return {
- "status": "error",
- "message": "CSV file is empty or has no data rows."
- }, previous_id, previous_id
-
- if total_rows > 100000:
- return {
- "status": "error",
- "message": f"File has {total_rows:,} rows. Maximum supported is 100,000 rows."
- }, previous_id, previous_id
-
- # Calculate chunks
- chunk_size = 5000
- total_chunks = math.ceil(total_rows / chunk_size)
-
- # Phase 2: Process chunks
- all_cards: list = []
- all_failed_rows: list = []
- chunks_processed = 0
- chunks_failed = 0
-
- for chunk_idx, cards, failed_rows in parse_arena_csv_chunked(filepath, chunk_size):
- # Update progress
- progress_pct = (chunk_idx + 1) / total_chunks
- processed_rows = min((chunk_idx + 1) * chunk_size, total_rows)
- progress(
- progress_pct,
- desc=f"Processing chunk {chunk_idx + 1}/{total_chunks} ({processed_rows:,}/{total_rows:,} rows)"
- )
-
- # Accumulate results
- all_cards.extend(cards)
- all_failed_rows.extend(failed_rows)
- chunks_processed += 1
-
- if len(failed_rows) > chunk_size * 0.5: # >50% failure rate
- chunks_failed += 1
-
- # Phase 3: Finalize
- progress(1.0, desc="Finalizing collection...")
-
- elapsed_time = time.time() - start_time
-
- # Create collection
- collection = Collection(
- name=f"Imported Collection ({total_rows:,} cards)",
- cards=all_cards,
- )
-
- # Generate collection ID (simple incrementing for now)
- collection_id = (previous_id or 0) + 1
-
- # Build result
- result = CollectionProcessingResult(
- collection_id=collection_id,
- total_cards=collection.total_cards,
- unique_cards=collection.unique_cards,
- total_quantity=sum(card.quantity for card in all_cards),
- chunks_processed=chunks_processed,
- chunks_failed=chunks_failed,
- failed_rows=all_failed_rows[:100], # Limit to first 100 failures
- processing_time_seconds=round(elapsed_time, 2),
- status="complete" if chunks_failed == 0 else "partial"
- )
-
- # Format response
- status = "success" if result.status == "complete" else "warning"
- response = {
- "status": status,
- "message": f"Collection processed successfully in {elapsed_time:.1f}s",
- "collection_id": collection_id,
- "statistics": {
- "total_rows": total_rows,
- "cards_imported": result.total_cards,
- "unique_cards": result.unique_cards,
- "total_quantity": result.total_quantity,
- "chunks_processed": chunks_processed,
- "chunks_failed": chunks_failed,
- "failed_rows_count": len(all_failed_rows),
- "processing_time": f"{elapsed_time:.2f}s"
- }
- }
-
- if all_failed_rows:
- response["failed_rows_sample"] = all_failed_rows[:20]
- response["message"] += f" ({len(all_failed_rows)} rows failed to parse)"
-
- return response, collection_id, collection_id
-
- except Exception as e:
- logger.exception(f"Collection upload error: {e}")
- return {
- "status": "error",
- "message": f"Failed to process collection: {str(e)}",
- "error_type": type(e).__name__
- }, previous_id, previous_id
-
-async def chat_streaming(message, history, deck_id):
- """Stream chat responses using ConcurrentChatService."""
- if not message or not message.strip():
- yield history or [], ""
- return
-
- history = history or []
-
- # Prepare context
- context = {}
- if deck_id:
- try:
- deck = await sql_service.get_deck(int(deck_id))
- if deck:
- analysis = await deck_analyzer.analyze_deck(deck)
- context = {
- "deck": {"name": deck.name, "format": deck.format},
- "analysis": {"overall_score": analysis.overall_score}
- }
- except Exception as e:
- logger.error(f"Failed to load deck context: {e}")
-
- # Call chat service
- # Note: ConcurrentChatService isn't a generator yet, so we await the full response
- # In a future update, we could make it stream
- try:
- result = await chat_service.chat(message, context)
- response_text = result["response"]
-
- if result.get("consensus_checked") and not result.get("consensus_passed"):
- response_text += f"\n\n⚠️ **Consensus Warning**: {result.get('consensus_breaker', {}).get('reason')}"
-
- # Gradio 6.x Chatbot uses messages as dictionaries with 'role' and 'content' keys
- history.append({"role": "user", "content": message})
- history.append({"role": "assistant", "content": response_text})
- yield history, ""
-
- except Exception as e:
- logger.exception("Chat failed")
- # Gradio 6.x Chatbot uses messages as dictionaries with 'role' and 'content' keys
- history.append({"role": "user", "content": message})
- history.append({"role": "assistant", "content": f"Error: {str(e)}"})
- yield history, ""
-
-# -----------------------------------------------------------------------------
-# UI Builders
-# -----------------------------------------------------------------------------
-
-REPO_URL = "https://github.com/clduab11/vawlrathh"
-HACKATHON_URL = "https://huggingface.co/MCP-1st-Birthday"
-HF_DEPLOYMENT_GUIDE_URL = f"{REPO_URL}/blob/main/docs/HF_DEPLOYMENT.md"
-
-def build_gpu_status_tab():
- """GPU status and initialization tab."""
- gr.Markdown("## GPU Status")
-
- gpu_status = gr.JSON(label="GPU Information", value={})
- init_btn = gr.Button("Initialize GPU", variant="primary")
-
- init_btn.click(
- fn=initialize_gpu,
- outputs=gpu_status
- )
-
- gr.Markdown(
- "Click 'Initialize GPU' to test GPU availability. "
- "This is optional - the app works on CPU if GPU is not available."
- )
-
-def build_deck_uploader_tab():
- """Deck uploader tab."""
- gr.Markdown("## Deck Uploads")
- deck_id_state = gr.State(value=None)
- deck_id_box = gr.Number(label="Latest Deck ID", interactive=False)
- upload_status = gr.JSON(label="Upload Response", value={})
-
- with gr.Row():
- csv_input = gr.File(file_types=[".csv"], label="Arena CSV Export")
- upload_btn = gr.Button("Upload CSV", variant="primary")
-
- upload_btn.click(
- fn=handle_csv_upload,
- inputs=[csv_input, deck_id_state],
- outputs=[upload_status, deck_id_state, deck_id_box],
- )
-
- gr.Markdown("### Arena Text Export")
- deck_text_input = gr.Textbox(
- lines=10,
- label="Arena Export",
- placeholder="4 Lightning Bolt (M11) 146\n2 Counterspell (MH2) 267",
- )
- format_dropdown = gr.Dropdown(
- choices=["Standard", "Pioneer", "Modern", "Alchemy", "Historic", "Explorer", "Timeless", "Brawl", "Historic Brawl", "Limited (Draft)", "Limited (Sealed)"],
- value="Standard",
- label="Format",
- )
- text_upload_btn = gr.Button("Upload Text", variant="secondary")
-
- text_upload_btn.click(
- fn=handle_text_upload,
- inputs=[deck_text_input, format_dropdown, deck_id_state],
- outputs=[upload_status, deck_id_state, deck_id_box],
- )
-
- gr.Markdown("### Tips")
- gr.Markdown(
- "* CSV uploads should come from the Steam Arena export.\n"
- "* Text uploads should be the Arena clipboard format.\n"
- "* The latest `deck_id` works across the Meta dashboard and chat tabs."
- )
-
-def build_collection_uploader_tab():
- """Build the Collection Upload tab UI for large-scale CSV imports."""
-
- gr.Markdown("""
- ## 📦 Collection Upload
-
- Import your entire MTG collection from a CSV file. This tab is optimized for large files
- with up to **70,000+ cards** using chunked processing.
-
- **Supported Format:** CSV with columns: `Quantity, Name, Set, Type, Mana Cost, CMC, Colors, Rarity`
- """)
-
- with gr.Row():
- with gr.Column(scale=2):
- collection_input = gr.File(
- file_types=[".csv"],
- label="Collection CSV File",
- elem_classes=["upload-zone", "upload-zone-large"]
- )
-
- with gr.Column(scale=1):
- upload_collection_btn = gr.Button(
- "📤 Upload Collection",
- variant="primary",
- size="lg"
- )
-
- # Status and results section
- with gr.Row():
- with gr.Column():
- collection_status = gr.JSON(
- label="Upload Status",
- value={"status": "ready", "message": "Select a CSV file to upload"}
- )
-
- with gr.Row():
- collection_id_box = gr.Number(
- label="Collection ID",
- interactive=False,
- value=None
- )
-
- # State for tracking
- collection_id_state = gr.State(value=None)
-
- # Tips section
- gr.Markdown("""
- ### 💡 Tips for Large Collections
-
- - **File Size:** Files up to 100MB are supported (approximately 70,000 cards)
- - **Processing:** Large files are processed in chunks of 5,000 rows
- - **Progress:** Watch the progress bar for real-time updates
- - **Timeout Prevention:** Progress updates keep the connection alive
- - **Partial Success:** If some rows fail, the rest will still be imported
-
- ### Expected CSV Format
-
- ```csv
- Quantity,Name,Set,Type,Mana Cost,CMC,Colors,Rarity
- 4,Lightning Bolt,M11,Instant,R,1,R,Common
- 2,Counterspell,MH2,Instant,UU,2,U,Uncommon
- 1,Black Lotus,LEA,Artifact,0,0,,Rare
- ```
- """)
-
- # Wire up the event handler
- upload_collection_btn.click(
- fn=handle_collection_upload,
- inputs=[collection_input, collection_id_state],
- outputs=[collection_status, collection_id_state, collection_id_box],
- show_progress="full"
- )
-
- return collection_input, upload_collection_btn, collection_status, collection_id_box, collection_id_state
-
-def build_analysis_tab():
- """Deck analysis and optimization tab."""
- gr.Markdown("## Deck Analysis & Optimization")
-
- deck_id_input = gr.Number(label="Deck ID", precision=0)
-
- with gr.Row():
- analyze_btn = gr.Button("Analyze Deck", variant="primary")
- optimize_btn = gr.Button("Get Suggestions", variant="secondary")
-
- analysis_json = gr.JSON(label="Analysis Results")
- suggestions_json = gr.JSON(label="Optimization Suggestions")
-
- analyze_btn.click(
- fn=handle_analyze,
- inputs=deck_id_input,
- outputs=analysis_json
- )
-
- optimize_btn.click(
- fn=handle_optimize,
- inputs=deck_id_input,
- outputs=suggestions_json
- )
-
-def build_chat_ui_tab():
- """Chat interface using Gradio Chatbot."""
- gr.Markdown("## Chat with Vawlrathh")
- gr.Markdown("Ask me about your deck, mulligan decisions, sideboard strategies, or meta matchups.")
-
- chatbot = gr.Chatbot(
- label="Live Conversation",
- height=400,
- )
- with gr.Row():
- msg = gr.Textbox(
- label="Message",
- placeholder="Ask Vawlrathh how to fix your deck...",
- scale=4,
- container=False,
- )
- submit_btn = gr.Button("Send", variant="primary", scale=1)
-
- with gr.Row():
- deck_context = gr.Number(label="Deck ID (optional)", precision=0, scale=2)
- clear = gr.Button("Clear Chat", variant="secondary", scale=1)
-
- # Wire up submit events
- msg.submit(
- fn=chat_streaming,
- inputs=[msg, chatbot, deck_context],
- outputs=[chatbot, msg],
- )
- submit_btn.click(
- fn=chat_streaming,
- inputs=[msg, chatbot, deck_context],
- outputs=[chatbot, msg],
- )
- clear.click(fn=lambda: ([], ""), inputs=None, outputs=[chatbot, msg], queue=False)
-
-def build_meta_dashboard_tab():
- """Meta dashboards tab."""
- gr.Markdown("## Meta Dashboards")
- format_dropdown = gr.Dropdown(
- choices=["Standard", "Pioneer", "Modern", "Alchemy", "Historic", "Explorer", "Timeless", "Brawl", "Historic Brawl", "Limited (Draft)", "Limited (Sealed)"],
- value="Standard",
- label="Format",
- )
- meta_btn = gr.Button("Load Meta Snapshot", variant="primary")
- meta_json = gr.JSON(label="Meta Intelligence", value={})
-
- meta_btn.click(
- fn=handle_meta_snapshot,
- inputs=format_dropdown,
- outputs=meta_json,
- )
-
- deck_input = gr.Number(label="Deck ID", precision=0)
- memory_btn = gr.Button("Load Smart Memory", variant="secondary")
- memory_json = gr.JSON(label="Memory Summary", value={})
-
- memory_btn.click(
- fn=handle_memory_summary,
- inputs=deck_input,
- outputs=memory_json,
- )
-
-def check_environment():
- """Check required environment variables and return HTML summary."""
- env_status = {}
- required_keys = {
- "OPENAI_API_KEY": "Required for AI-powered deck analysis and chat",
- "ANTHROPIC_API_KEY": "Required for consensus checking",
- }
- optional_keys = {
- "HF_TOKEN": "Used for CLI-based syncs and GitHub workflow dispatch",
- "TAVILY_API_KEY": "Recommended for meta intelligence",
- "EXA_API_KEY": "Recommended for semantic search",
- }
-
- has_missing_required = False
-
- for key, description in required_keys.items():
- if os.getenv(key):
- env_status[key] = "✓ Configured"
- else:
- env_status[key] = f"✗ Missing - {description}"
- has_missing_required = True
-
- for key, description in optional_keys.items():
- if os.getenv(key):
- env_status[key] = "✓ Configured"
- else:
- env_status[key] = f"⚠ Not configured - {description}"
-
- status_html = "Environment Configuration
"
- for key, status in env_status.items():
- status_html += f"- {key}: {status}
"
- status_html += "
"
-
- if has_missing_required:
- status_html += (
- "⚠ Warning: "
- "Some required API keys are missing. Configure them in the HF Space settings."
- "
"
- )
-
- return status_html
-
-def create_gradio_interface():
- """Create the main Gradio interface."""
-
- # About content
- about_html = textwrap.dedent(
- f"""
-
-
Vawlrathh
-
- "Your deck's terrible. Let me show you how to fix it."
- — Vawlrathh, The Small'n
-
-
-
🎯 What This Is
-
- Listen up. I'm Vawlrathh, The Small'n—a pint-sized,
- sharp-tongued version of Volrath, The Fallen. Despite my stature, I
- know MTG Arena better than you know your own deck (which, frankly,
- isn't saying much).
-
-
-
- Vawlrathh is an MCP-powered deck analysis tool
- that actually works. It analyzes your janky brews, tells you what's
- wrong (plenty), and helps you build something that won't embarrass
- you at FNM.
-
-
-
What Makes This Not-Garbage
-
- - Real-Time Strategy Chat: Talk to me directly.
- I'll tell you the truth
- - AI Consensus Checking: Two AI brains so you don't
- get bad advice
- - Sequential Reasoning: Breaks down complex
- decisions into steps you can follow
- - Full MCP Integration: Memory, sequential thinking,
- omnisearch—the works
-
-
-
📖 How to Use
-
-
Uploading Decks
-
- - Export from Arena: In MTG Arena, go to your deck and click "Export" to copy to clipboard
- - Paste in Deck Uploads tab: Go to the "Deck Uploads" tab and paste your deck list in the text area
- - Or upload CSV: If you have a CSV export from Steam Arena, upload it directly
- - Select format: Choose your format (Standard, Pioneer, Modern, etc.) and click "Upload"
- - Note the Deck ID: After upload, you'll receive a Deck ID to use in other tabs
-
-
-
Using Chat
-
- - Go to Chat tab: Navigate to the "Chat" tab
- - Enter Deck ID (optional): If you want deck-specific advice, enter your Deck ID
- - Ask anything: Ask for deck analysis, mulligan advice, sideboard strategies, or meta matchup tips
- - Example questions:
-
- - "What are my deck's weaknesses?"
- - "Should I mulligan a hand with 2 lands and 5 spells?"
- - "How do I sideboard against control?"
- - "What cards should I cut to improve consistency?"
-
-
-
-
-
Navigating Tabs
-
- - Deck Uploads: Upload your deck via CSV or text paste
- - Analysis: Get detailed deck analysis including mana curve, strengths, weaknesses, and synergies. Enter your Deck ID and click "Analyze Deck" for full breakdown, or "Get Suggestions" for optimization recommendations
- - Chat: Have a conversation with Vawlrathh about strategy, mulligan decisions, and deck improvements
- - Meta Intelligence: View current meta snapshots for your format and track deck performance history
- - Status: Check API key configuration status
- - GPU Status: Test GPU availability (optional - app works on CPU)
-
-
-
🎖️ MCP 1st Birthday Hackathon
-
- This project is submitted for the
- MCP 1st Birthday Hackathon. Visit the
-
- hackathon page
-
- to see more MCP-powered projects.
-
-
-
- Repository:
-
- github.com/clduab11/vawlrathh
-
-
-
- Diminutive in size, not in strategic prowess. |
-
- GitHub Repository
-
- |
-
- MCP 1st Birthday Hackathon
-
-
- """
- )
- gr.Markdown(footer_md)
-
- return interface
-
-# Create the Gradio app
-demo = create_gradio_interface()
-
-# Launch logic not needed for HF Spaces, as it imports 'demo' directly
-# via the sdk: gradio configuration.
-if __name__ == "__main__":
- demo.launch()
+"""Hugging Face Space wrapper for Vawlrathh.
+
+This module provides a pure Gradio application for deployment on Hugging Face Spaces
+with ZeroGPU support. It integrates directly with Vawlrathh services instead of
+using a separate FastAPI backend, ensuring compatibility with the Gradio SDK runner.
+
+"Your deck's terrible. Let me show you how to fix it."
+— Vawlrathh, The Small'n
+"""
+
+import logging
+import math
+import os
+import sys
+import textwrap
+import time
+import json
+import asyncio
+from dataclasses import dataclass
+from typing import Any, Callable, Dict, List, Optional, Tuple
+
+from dotenv import load_dotenv
+import gradio as gr
+import spaces
+
+# Load environment variables from .env file
+load_dotenv()
+
+# Ensure project root is in python path
+sys.path.append(os.path.dirname(os.path.abspath(__file__)))
+
+# Import Services
+from src.services.smart_sql import SmartSQLService
+from src.services.deck_analyzer import DeckAnalyzer
+from src.services.collection_analyzer import CollectionAnalyzer
+from src.services.meta_intelligence import MetaIntelligenceService
+from src.services.chat_agent import ConcurrentChatService
+from src.services.smart_inference import SmartInferenceService
+from src.services.card_market_service import CardMarketService
+from src.utils.csv_parser import parse_arena_csv, parse_deck_string
+
+# Configure logging
+logging.basicConfig(
+ level=logging.INFO,
+ format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+# -----------------------------------------------------------------------------
+# Service Initialization
+# -----------------------------------------------------------------------------
+
+# Initialize services
+# Note: Async initialization happens in the event loop
+sql_service = SmartSQLService()
+meta_service = MetaIntelligenceService()
+deck_analyzer = DeckAnalyzer(meta_service=meta_service)
+collection_analyzer = CollectionAnalyzer()
+chat_service = ConcurrentChatService()
+inference_service = SmartInferenceService()
+card_market_service = CardMarketService()
+
+# -----------------------------------------------------------------------------
+# GPU / Spaces Configuration
+# -----------------------------------------------------------------------------
+
+@spaces.GPU(duration=10)
+def initialize_gpu():
+ """Initialize GPU runtime for HF Spaces ZERO.
+
+ This function exists primarily to satisfy the ZeroGPU requirement that
+ at least one function must be decorated with @spaces.GPU.
+ """
+ import torch
+ if torch.cuda.is_available():
+ device = torch.cuda.get_device_name(0)
+ logger.info(f"GPU initialized: {device}")
+ return {"gpu": device, "cuda_available": True}
+ return {"gpu": None, "cuda_available": False}
+
+# -----------------------------------------------------------------------------
+# Helper Functions (Direct Service Calls)
+# -----------------------------------------------------------------------------
+
+async def ensure_db_initialized():
+ """Ensure database is initialized."""
+ await sql_service.init_db()
+
+async def handle_csv_upload(uploaded_file, previous_id):
+ """Handle CSV upload by calling services directly."""
+ if not uploaded_file:
+ return {"status": "error", "message": "No CSV file selected"}, previous_id, previous_id
+
+ try:
+ # Ensure DB is ready
+ await ensure_db_initialized()
+
+ # Handle both file path (Gradio 4.x+) and file object (older Gradio)
+ if hasattr(uploaded_file, 'read'):
+ content = uploaded_file.read()
+ else:
+ # uploaded_file is a path string (NamedString or str)
+ with open(str(uploaded_file), 'r', encoding='utf-8') as f:
+ content = f.read()
+ if isinstance(content, bytes):
+ content = content.decode('utf-8')
+
+ deck = parse_arena_csv(content)
+ deck_id = await sql_service.store_deck(deck)
+
+ return {
+ "status": "success",
+ "message": f"Deck '{deck.name}' uploaded successfully",
+ "deck_id": deck_id
+ }, deck_id, deck_id
+ except Exception as e:
+ logger.exception("CSV upload failed")
+ return {"status": "error", "message": str(e)}, previous_id, previous_id
+
+async def handle_text_upload(deck_text, fmt, previous_id):
+ """Handle text upload by calling services directly."""
+ if not deck_text or not deck_text.strip():
+ return {"status": "error", "message": "Deck text is empty"}, previous_id, previous_id
+
+ try:
+ # Ensure DB is ready
+ await ensure_db_initialized()
+
+ deck = parse_deck_string(deck_text)
+ deck.format = fmt
+ deck_id = await sql_service.store_deck(deck)
+
+ return {
+ "status": "success",
+ "message": f"Deck uploaded successfully",
+ "deck_id": deck_id
+ }, deck_id, deck_id
+ except Exception as e:
+ logger.exception("Text upload failed")
+ return {"status": "error", "message": str(e)}, previous_id, previous_id
+
+async def handle_analyze(deck_id: float):
+ """Analyze a deck using DeckAnalyzer."""
+ if not deck_id:
+ return {"status": "error", "message": "No deck ID provided"}
+
+ try:
+ deck = await sql_service.get_deck(int(deck_id))
+ if not deck:
+ return {"status": "error", "message": "Deck not found"}
+
+ analysis = await deck_analyzer.analyze_deck(deck)
+
+ # Convert to dict for JSON output
+ return {
+ "deck_name": analysis.deck_name,
+ "overall_score": analysis.overall_score,
+ "mana_curve": analysis.mana_curve.__dict__,
+ "strengths": analysis.strengths,
+ "weaknesses": analysis.weaknesses,
+ "synergies": [s.__dict__ for s in analysis.synergies],
+ "meta_matchups": [m.__dict__ for m in analysis.meta_matchups]
+ }
+ except Exception as e:
+ logger.exception("Analysis failed")
+ return {"status": "error", "message": str(e)}
+
+async def handle_optimize(deck_id: float):
+ """Generate optimization suggestions."""
+ if not deck_id:
+ return {"status": "error", "message": "No deck ID provided"}
+
+ try:
+ deck = await sql_service.get_deck(int(deck_id))
+ if not deck:
+ return {"status": "error", "message": "Deck not found"}
+
+ analysis = await deck_analyzer.analyze_deck(deck)
+ suggestions = await inference_service.generate_suggestions(deck, analysis)
+
+ return [s.__dict__ for s in suggestions]
+ except Exception as e:
+ logger.exception("Optimization failed")
+ return {"status": "error", "message": str(e)}
+
+async def handle_meta_snapshot(game_format: str):
+ """Fetch meta snapshot."""
+ try:
+ snapshot = await meta_service.get_current_meta(game_format)
+ return meta_service.to_dict(snapshot)
+ except Exception as e:
+ logger.exception("Meta snapshot failed")
+ return {"status": "error", "message": str(e)}
+
+async def handle_memory_summary(deck_id: float):
+ """Fetch deck performance history."""
+ if not deck_id:
+ return {"status": "error", "message": "No deck ID provided"}
+
+ try:
+ history = await sql_service.get_deck_performance(int(deck_id))
+ return history
+ except Exception as e:
+ logger.exception("Memory summary failed")
+ return {"status": "error", "message": str(e)}
+
+async def handle_collection_upload(
+ uploaded_file,
+ previous_id,
+ progress=gr.Progress(track_tqdm=True)
+):
+ """
+ Handle large collection CSV upload with chunked processing and progress tracking.
+
+ Args:
+ uploaded_file: File path from gr.File component
+ previous_id: Previous collection ID from state
+ progress: Gradio progress tracker
+
+ Returns:
+ Tuple of (status_dict, collection_id, collection_id)
+ """
+ from src.utils.csv_parser import count_csv_rows, parse_arena_csv_chunked
+ from src.models.deck import Collection, CollectionProcessingResult, Card
+
+ start_time = time.time()
+
+ # Validate input
+ if uploaded_file is None:
+ return {
+ "status": "error",
+ "message": "No file uploaded. Please select a CSV file."
+ }, previous_id, previous_id
+
+ # Get file path (Gradio 4.x+ returns path string)
+ filepath = uploaded_file if isinstance(uploaded_file, str) else uploaded_file.name
+
+ try:
+ # Phase 1: Count rows for progress calculation
+ progress(0, desc="Analyzing file size...")
+ total_rows = count_csv_rows(filepath)
+
+ if total_rows == 0:
+ return {
+ "status": "error",
+ "message": "CSV file is empty or has no data rows."
+ }, previous_id, previous_id
+
+ if total_rows > 100000:
+ return {
+ "status": "error",
+ "message": f"File has {total_rows:,} rows. Maximum supported is 100,000 rows."
+ }, previous_id, previous_id
+
+ # Calculate chunks
+ chunk_size = 5000
+ total_chunks = math.ceil(total_rows / chunk_size)
+
+ # Phase 2: Process chunks
+ all_cards: list = []
+ all_failed_rows: list = []
+ chunks_processed = 0
+ chunks_failed = 0
+
+ for chunk_idx, cards, failed_rows in parse_arena_csv_chunked(filepath, chunk_size):
+ # Update progress
+ progress_pct = (chunk_idx + 1) / total_chunks
+ processed_rows = min((chunk_idx + 1) * chunk_size, total_rows)
+ progress(
+ progress_pct,
+ desc=f"Processing chunk {chunk_idx + 1}/{total_chunks} ({processed_rows:,}/{total_rows:,} rows)"
+ )
+
+ # Accumulate results
+ all_cards.extend(cards)
+ all_failed_rows.extend(failed_rows)
+ chunks_processed += 1
+
+ if len(failed_rows) > chunk_size * 0.5: # >50% failure rate
+ chunks_failed += 1
+
+ # Phase 3: Finalize
+ progress(1.0, desc="Finalizing collection...")
+
+ elapsed_time = time.time() - start_time
+
+ # Create collection
+ collection = Collection(
+ name=f"Imported Collection ({total_rows:,} cards)",
+ cards=all_cards,
+ )
+
+ # Perform analysis immediately
+ analysis = await collection_analyzer.analyze_collection(collection)
+
+ # Generate collection ID (simple incrementing for now)
+ collection_id = (previous_id or 0) + 1
+
+ # Build result
+ result = CollectionProcessingResult(
+ collection_id=collection_id,
+ total_cards=collection.total_cards,
+ unique_cards=collection.unique_cards,
+ total_quantity=sum(card.quantity for card in all_cards),
+ chunks_processed=chunks_processed,
+ chunks_failed=chunks_failed,
+ failed_rows=all_failed_rows[:100], # Limit to first 100 failures
+ processing_time_seconds=round(elapsed_time, 2),
+ status="complete" if chunks_failed == 0 else "partial"
+ )
+
+ # Format response
+ status = "success" if result.status == "complete" else "warning"
+ response = {
+ "status": status,
+ "message": f"Collection processed successfully in {elapsed_time:.1f}s",
+ "collection_id": collection_id,
+ "statistics": {
+ "total_rows": total_rows,
+ "cards_imported": result.total_cards,
+ "unique_cards": result.unique_cards,
+ "total_quantity": result.total_quantity,
+ "chunks_processed": chunks_processed,
+ "chunks_failed": chunks_failed,
+ "failed_rows_count": len(all_failed_rows),
+ "processing_time": f"{elapsed_time:.2f}s"
+ },
+ "analysis": {
+ "rarity_distribution": analysis.rarity_distribution,
+ "color_distribution": analysis.color_distribution,
+ "set_distribution": dict(list(analysis.set_distribution.items())[:10]), # Top 10 sets
+ "completion_score": analysis.completion_score
+ }
+ }
+
+ if all_failed_rows:
+ response["failed_rows_sample"] = all_failed_rows[:20]
+ response["message"] += f" ({len(all_failed_rows)} rows failed to parse)"
+
+ return response, collection_id, collection_id
+
+ except Exception as e:
+ logger.exception(f"Collection upload error: {e}")
+ return {
+ "status": "error",
+ "message": f"Failed to process collection: {str(e)}",
+ "error_type": type(e).__name__
+ }, previous_id, previous_id
+
+async def handle_collection_analyze(collection_id: float):
+ """Analyze a collection."""
+ if not collection_id:
+ return {"status": "error", "message": "No collection ID provided"}
+
+ try:
+ # In a real app with persistence, we'd load from DB.
+ # Since we don't have a 'get_collection' in SQL service yet and state is ephemeral in this demo,
+ # we'll mock the retrieval or need to pass the collection object.
+ # For this hackathon demo, we'll assume the collection was just uploaded and we might need to
+ # re-parse or store it in memory.
+ # To keep it simple and robust without DB schema changes:
+ # We will return a placeholder message or implement a simple in-memory cache if needed.
+ # BUT, looking at handle_collection_upload, it returns an ID but doesn't persist to SQL (it seems).
+ # Let's check handle_collection_upload again... it creates a Collection object but doesn't save it to sql_service?
+ # Ah, handle_collection_upload just returns stats.
+
+ # To make this work properly, we should probably return the analysis directly from upload
+ # OR implement persistence. Given the constraints, let's update handle_collection_upload to
+ # return the analysis immediately or store in a global cache (not ideal for multi-user but ok for demo).
+
+ # BETTER APPROACH: Let's just return a "Not Implemented - Persistence Required"
+ # OR, since the user asked for "Collection Analysis", let's make the upload button DO the analysis.
+
+ return {"status": "info", "message": "Analysis requires database persistence (Coming Soon)"}
+
+ except Exception as e:
+ logger.exception("Collection analysis failed")
+ return {"status": "error", "message": str(e)}
+
+async def chat_streaming(message, history, deck_id):
+ """Stream chat responses using ConcurrentChatService."""
+ if not message or not message.strip():
+ yield history or [], ""
+ return
+
+ history = history or []
+
+ # Prepare context
+ context = {}
+ if deck_id:
+ try:
+ deck = await sql_service.get_deck(int(deck_id))
+ if deck:
+ analysis = await deck_analyzer.analyze_deck(deck)
+ context = {
+ "deck": {"name": deck.name, "format": deck.format},
+ "analysis": {"overall_score": analysis.overall_score}
+ }
+ except Exception as e:
+ logger.error(f"Failed to load deck context: {e}")
+
+ # Call chat service
+ # Note: ConcurrentChatService isn't a generator yet, so we await the full response
+ # In a future update, we could make it stream
+ try:
+ result = await chat_service.chat(message, context)
+ response_text = result["response"]
+
+ if result.get("consensus_checked") and not result.get("consensus_passed"):
+ response_text += f"\n\n⚠️ **Consensus Warning**: {result.get('consensus_breaker', {}).get('reason')}"
+
+ # Gradio 6.x Chatbot uses messages as dictionaries with 'role' and 'content' keys
+ history.append({"role": "user", "content": message})
+ history.append({"role": "assistant", "content": response_text})
+ yield history, ""
+
+ except Exception as e:
+ logger.exception("Chat failed")
+ # Gradio 6.x Chatbot uses messages as dictionaries with 'role' and 'content' keys
+ history.append({"role": "user", "content": message})
+ history.append({"role": "assistant", "content": f"Error: {str(e)}"})
+ yield history, ""
+
+# -----------------------------------------------------------------------------
+# UI Builders
+# -----------------------------------------------------------------------------
+
+REPO_URL = "https://github.com/clduab11/vawlrathh"
+HACKATHON_URL = "https://huggingface.co/MCP-1st-Birthday"
+HF_DEPLOYMENT_GUIDE_URL = f"{REPO_URL}/blob/main/docs/HF_DEPLOYMENT.md"
+
+def build_gpu_status_tab():
+ """GPU status and initialization tab."""
+ gr.Markdown("## GPU Status")
+
+ gpu_status = gr.JSON(label="GPU Information", value={})
+ init_btn = gr.Button("Initialize GPU", variant="primary")
+
+ init_btn.click(
+ fn=initialize_gpu,
+ outputs=gpu_status
+ )
+
+ gr.Markdown(
+ "Click 'Initialize GPU' to test GPU availability. "
+ "This is optional - the app works on CPU if GPU is not available."
+ )
+
+def build_deck_uploader_tab():
+ """Deck uploader tab."""
+ gr.Markdown("## Deck Uploads")
+ deck_id_state = gr.State(value=None)
+ deck_id_box = gr.Number(label="Latest Deck ID", interactive=False)
+ upload_status = gr.JSON(label="Upload Response", value={})
+
+ with gr.Row():
+ csv_input = gr.File(file_types=[".csv"], label="Arena CSV Export")
+ upload_btn = gr.Button("Upload CSV", variant="primary")
+
+ upload_btn.click(
+ fn=handle_csv_upload,
+ inputs=[csv_input, deck_id_state],
+ outputs=[upload_status, deck_id_state, deck_id_box],
+ )
+
+ gr.Markdown("### Arena Text Export")
+ deck_text_input = gr.Textbox(
+ lines=10,
+ label="Arena Export",
+ placeholder="4 Lightning Bolt (M11) 146\n2 Counterspell (MH2) 267",
+ )
+ format_dropdown = gr.Dropdown(
+ choices=["Standard", "Pioneer", "Modern", "Alchemy", "Historic", "Explorer", "Timeless", "Brawl", "Historic Brawl", "Limited (Draft)", "Limited (Sealed)"],
+ value="Standard",
+ label="Format",
+ )
+ text_upload_btn = gr.Button("Upload Text", variant="secondary")
+
+ text_upload_btn.click(
+ fn=handle_text_upload,
+ inputs=[deck_text_input, format_dropdown, deck_id_state],
+ outputs=[upload_status, deck_id_state, deck_id_box],
+ )
+
+ gr.Markdown("### Tips")
+ gr.Markdown(
+ "* CSV uploads should come from the Steam Arena export.\n"
+ "* Text uploads should be the Arena clipboard format.\n"
+ "* The latest `deck_id` works across the Meta dashboard and chat tabs."
+ )
+
+def build_collection_uploader_tab():
+ """Build the Collection Upload tab UI for large-scale CSV imports."""
+
+ gr.Markdown("""
+ ## 📦 Collection Upload
+
+ Import your entire MTG collection from a CSV file. This tab is optimized for large files
+ with up to **70,000+ cards** using chunked processing.
+
+ **Supported Format:** CSV with columns: `Quantity, Name, Set, Type, Mana Cost, CMC, Colors, Rarity`
+ """)
+
+ with gr.Row():
+ with gr.Column(scale=2):
+ collection_input = gr.File(
+ file_types=[".csv"],
+ label="Collection CSV File",
+ elem_classes=["upload-zone", "upload-zone-large"]
+ )
+
+ with gr.Column(scale=1):
+ upload_collection_btn = gr.Button(
+ "📤 Upload Collection",
+ variant="primary",
+ size="lg"
+ )
+
+ # Status and results section
+ with gr.Row():
+ with gr.Column():
+ collection_status = gr.JSON(
+ label="Analysis Results",
+ value={"status": "ready", "message": "Select a CSV file to upload (Arena or Untapped.gg format)"}
+ )
+
+ with gr.Row():
+ collection_id_box = gr.Number(
+ label="Collection ID",
+ interactive=False,
+ value=None
+ )
+
+ # State for tracking
+ collection_id_state = gr.State(value=None)
+
+ # Tips section
+ gr.Markdown("""
+ ### 💡 Tips for Large Collections
+
+ - **File Size:** Files up to 100MB are supported (approximately 70,000 cards)
+ - **Processing:** Large files are processed in chunks of 5,000 rows
+ - **Progress:** Watch the progress bar for real-time updates
+ - **Timeout Prevention:** Progress updates keep the connection alive
+ - **Partial Success:** If some rows fail, the rest will still be imported
+
+ ### Expected CSV Format
+
+ **Steam Arena:**
+ ```csv
+ Quantity,Name,Set,Type,Mana Cost,CMC,Colors,Rarity
+ 4,Lightning Bolt,M11,Instant,R,1,R,Common
+ ```
+
+ **Untapped.gg:**
+ ```csv
+ Id #,Name,Set,Color,Rarity,Quantity
+ 123,Lightning Bolt,M11,Red,Common,4
+ ```
+ """)
+
+ # Wire up the event handler
+ upload_collection_btn.click(
+ fn=handle_collection_upload,
+ inputs=[collection_input, collection_id_state],
+ outputs=[collection_status, collection_id_state, collection_id_box],
+ show_progress="full"
+ )
+
+ return collection_input, upload_collection_btn, collection_status, collection_id_box, collection_id_state
+
+def build_analysis_tab():
+ """Deck analysis and optimization tab."""
+ gr.Markdown("## Deck Analysis & Optimization")
+
+ deck_id_input = gr.Number(label="Deck ID", precision=0)
+
+ with gr.Row():
+ analyze_btn = gr.Button("Analyze Deck", variant="primary")
+ optimize_btn = gr.Button("Get Suggestions", variant="secondary")
+
+ analysis_json = gr.JSON(label="Analysis Results")
+ suggestions_json = gr.JSON(label="Optimization Suggestions")
+
+ analyze_btn.click(
+ fn=handle_analyze,
+ inputs=deck_id_input,
+ outputs=analysis_json
+ )
+
+ optimize_btn.click(
+ fn=handle_optimize,
+ inputs=deck_id_input,
+ outputs=suggestions_json
+ )
+
+def build_chat_ui_tab():
+ """Chat interface using Gradio Chatbot."""
+ gr.Markdown("## Chat with Vawlrathh")
+ gr.Markdown("Ask me about your deck, mulligan decisions, sideboard strategies, or meta matchups.")
+
+ chatbot = gr.Chatbot(
+ label="Live Conversation",
+ height=400,
+ )
+ with gr.Row():
+ msg = gr.Textbox(
+ label="Message",
+ placeholder="Ask Vawlrathh how to fix your deck...",
+ scale=4,
+ container=False,
+ )
+ submit_btn = gr.Button("Send", variant="primary", scale=1)
+
+ with gr.Row():
+ deck_context = gr.Number(label="Deck ID (optional)", precision=0, scale=2)
+ clear = gr.Button("Clear Chat", variant="secondary", scale=1)
+
+ # Wire up submit events
+ msg.submit(
+ fn=chat_streaming,
+ inputs=[msg, chatbot, deck_context],
+ outputs=[chatbot, msg],
+ )
+ submit_btn.click(
+ fn=chat_streaming,
+ inputs=[msg, chatbot, deck_context],
+ outputs=[chatbot, msg],
+ )
+ clear.click(fn=lambda: ([], ""), inputs=None, outputs=[chatbot, msg], queue=False)
+
+def build_meta_dashboard_tab():
+ """Meta dashboards tab."""
+ gr.Markdown("## Meta Dashboards")
+ format_dropdown = gr.Dropdown(
+ choices=["Standard", "Pioneer", "Modern", "Alchemy", "Historic", "Explorer", "Timeless", "Brawl", "Historic Brawl", "Limited (Draft)", "Limited (Sealed)"],
+ value="Standard",
+ label="Format",
+ )
+ meta_btn = gr.Button("Load Meta Snapshot", variant="primary")
+ meta_json = gr.JSON(label="Meta Intelligence", value={})
+
+ meta_btn.click(
+ fn=handle_meta_snapshot,
+ inputs=format_dropdown,
+ outputs=meta_json,
+ )
+
+ deck_input = gr.Number(label="Deck ID", precision=0)
+ memory_btn = gr.Button("Load Smart Memory", variant="secondary")
+ memory_json = gr.JSON(label="Memory Summary", value={})
+
+ memory_btn.click(
+ fn=handle_memory_summary,
+ inputs=deck_input,
+ outputs=memory_json,
+ )
+
+def check_environment():
+ """Check required environment variables and return HTML summary."""
+ env_status = {}
+ required_keys = {
+ "OPENAI_API_KEY": "Required for AI-powered deck analysis and chat",
+ "ANTHROPIC_API_KEY": "Required for consensus checking",
+ }
+ optional_keys = {
+ "HF_TOKEN": "Used for CLI-based syncs and GitHub workflow dispatch",
+ "TAVILY_API_KEY": "Recommended for meta intelligence",
+ "EXA_API_KEY": "Recommended for semantic search",
+ }
+
+ has_missing_required = False
+
+ for key, description in required_keys.items():
+ if os.getenv(key):
+ env_status[key] = "✓ Configured"
+ else:
+ env_status[key] = f"✗ Missing - {description}"
+ has_missing_required = True
+
+ for key, description in optional_keys.items():
+ if os.getenv(key):
+ env_status[key] = "✓ Configured"
+ else:
+ env_status[key] = f"⚠ Not configured - {description}"
+
+ status_html = "Environment Configuration
"
+ for key, status in env_status.items():
+ status_html += f"- {key}: {status}
"
+ status_html += "
"
+
+ if has_missing_required:
+ status_html += (
+ "⚠ Warning: "
+ "Some required API keys are missing. Configure them in the HF Space settings."
+ "
"
+ )
+
+ return status_html
+
+def create_gradio_interface():
+ """Create the main Gradio interface."""
+
+ # About content
+ about_html = textwrap.dedent(
+ f"""
+
+
Vawlrathh
+
+ "Your deck's terrible. Let me show you how to fix it."
+ — Vawlrathh, The Small'n
+
+
+
🎯 What This Is
+
+ Listen up. I'm Vawlrathh, The Small'n—a pint-sized,
+ sharp-tongued version of Volrath, The Fallen. Despite my stature, I
+ know MTG Arena better than you know your own deck (which, frankly,
+ isn't saying much).
+
+
+
+ Vawlrathh is an MCP-powered deck analysis tool
+ that actually works. It analyzes your janky brews, tells you what's
+ wrong (plenty), and helps you build something that won't embarrass
+ you at FNM.
+
+
+
What Makes This Not-Garbage
+
+ - Real-Time Strategy Chat: Talk to me directly.
+ I'll tell you the truth
+ - AI Consensus Checking: Two AI brains so you don't
+ get bad advice
+ - Sequential Reasoning: Breaks down complex
+ decisions into steps you can follow
+ - Full MCP Integration: Memory, sequential thinking,
+ omnisearch—the works
+
+
+
📖 How to Use
+
+
Uploading Decks
+
+ - Export from Arena: In MTG Arena, go to your deck and click "Export" to copy to clipboard
+ - Paste in Deck Uploads tab: Go to the "Deck Uploads" tab and paste your deck list in the text area
+ - Or upload CSV: If you have a CSV export from Steam Arena, upload it directly
+ - Select format: Choose your format (Standard, Pioneer, Modern, etc.) and click "Upload"
+ - Note the Deck ID: After upload, you'll receive a Deck ID to use in other tabs
+
+
+
Using Chat
+
+ - Go to Chat tab: Navigate to the "Chat" tab
+ - Enter Deck ID (optional): If you want deck-specific advice, enter your Deck ID
+ - Ask anything: Ask for deck analysis, mulligan advice, sideboard strategies, or meta matchup tips
+ - Example questions:
+
+ - "What are my deck's weaknesses?"
+ - "Should I mulligan a hand with 2 lands and 5 spells?"
+ - "How do I sideboard against control?"
+ - "What cards should I cut to improve consistency?"
+
+
+
+
+
Navigating Tabs
+
+ - Deck Uploads: Upload your deck via CSV or text paste
+ - Analysis: Get detailed deck analysis including mana curve, strengths, weaknesses, and synergies. Enter your Deck ID and click "Analyze Deck" for full breakdown, or "Get Suggestions" for optimization recommendations
+ - Chat: Have a conversation with Vawlrathh about strategy, mulligan decisions, and deck improvements
+ - Meta Intelligence: View current meta snapshots for your format and track deck performance history
+ - Status: Check API key configuration status
+ - GPU Status: Test GPU availability (optional - app works on CPU)
+
+
+
🎖️ MCP 1st Birthday Hackathon
+
+ This project is submitted for the
+ MCP 1st Birthday Hackathon. Visit the
+
+ hackathon page
+
+ to see more MCP-powered projects.
+
+
+
+ Repository:
+
+ github.com/clduab11/vawlrathh
+
+
+
+ Diminutive in size, not in strategic prowess. |
+
+ GitHub Repository
+
+ |
+
+ MCP 1st Birthday Hackathon
+
+
+ """
+ )
+ gr.Markdown(footer_md)
+
+ return interface
+
+# Create the Gradio app
+demo = create_gradio_interface()
+
+# Launch logic not needed for HF Spaces, as it imports 'demo' directly
+# via the sdk: gradio configuration.
+if __name__ == "__main__":
+ demo.launch()