app.py

import gradio as gr
from gradio_chessboard import Chessboard
import chess
import chess.engine
import pandas as pd
import os
import uuid
from openai import OpenAI
import time

SYSTEM_KEY = os.getenv("OPENAI_API_KEY")
STOCKFISH_PATH = "/usr/games/stockfish"

def get_client(user_key):
    key = user_key.strip() if user_key else SYSTEM_KEY
    if not key: return None
    return OpenAI(api_key=key)

def _load_lichess_openings(path_prefix="/app/data/lichess_openings/dist/"):
    try:
        files = [f"{path_prefix}{vol}.tsv" for vol in ("a", "b", "c", "d", "e")]
        dfs = []
        for fn in files:
            if os.path.exists(fn):
                df = pd.read_csv(fn, sep="\t", usecols=["eco", "name", "pgn", "uci", "epd"])
                dfs.append(df)
        if not dfs: return pd.DataFrame()
        return pd.concat(dfs, ignore_index=True)
    except:
        return pd.DataFrame()

OPENINGS_DB = _load_lichess_openings()

def analyze_tactics(board):
    fen = board.fen()
    pins = []
    turn = board.turn
    for sq in chess.SQUARES:
        piece = board.piece_at(sq)
        if piece and piece.color == turn:
            if board.is_pinned(turn, sq):
                pins.append(chess.square_name(sq))
    
    check = board.is_check()
    values = {chess.PAWN: 1, chess.KNIGHT: 3, chess.BISHOP: 3, chess.ROOK: 5, chess.QUEEN: 9}
    w_mat = sum(len(board.pieces(pt, chess.WHITE)) * val for pt, val in values.items())
    b_mat = sum(len(board.pieces(pt, chess.BLACK)) * val for pt, val in values.items())
    
    analysis = []
    if check: analysis.append("KING IN CHECK.")
    if pins: analysis.append(f"Pinned pieces at: {', '.join(pins)}.")
    analysis.append(f"Material: White {w_mat} vs Black {b_mat}.")
    return " ".join(analysis)

def tool_engine_analysis(board, time_limit=0.5):
    if board.is_game_over(): return "GAME OVER", "NONE", "NONE"
    if not os.path.exists(STOCKFISH_PATH): return "0", "Engine Missing", ""
    
    try:
        with chess.engine.SimpleEngine.popen_uci(STOCKFISH_PATH) as engine:
            info = engine.analyse(board, chess.engine.Limit(time=time_limit))
            score = info["score"].white()
            
            if score.is_mate():
                score_val = f"MATE IN {abs(score.mate())}"
            else:
                score_val = str(score.score())

            best_move = info.get("pv", [None])[0]
            if best_move:
                origin = chess.square_name(best_move.from_square)
                dest = chess.square_name(best_move.to_square)
                move_uci = f"{origin} -> {dest}"
            else:
                move_uci = "NO MOVE"

            return score_val, move_uci, move_uci
    except:
        return "N/A", "ANALYSIS ERROR", ""

def tool_ai_play(board, level):
    levels = {
        "Beginner": {"time": 0.01, "skill": 1, "depth": 1},
        "Intermediate": {"time": 0.1, "skill": 8, "depth": 6},
        "Advanced": {"time": 0.5, "skill": 15, "depth": 12},
        "Grandmaster": {"time": 1.0, "skill": 20, "depth": 18}
    }
    config = levels.get(level, levels["Beginner"])
    try:
        with chess.engine.SimpleEngine.popen_uci(STOCKFISH_PATH) as engine:
            engine.configure({"Skill Level": config["skill"]})
            result = engine.play(board, chess.engine.Limit(time=config["time"], depth=config["depth"]))
            return result.move
    except:
        return None

# --- AUDIO & TRANSCRIPTION ---
def generate_voice(client, text):
    if not client or not text: return None
    try:
        response = client.audio.speech.create(
            model="tts-1", voice="onyx", input=text, speed=1.15
        )
        unique = f"deepblue_{uuid.uuid4()}.mp3"
        path = os.path.join("/tmp", unique)
        with open(path, "wb") as f:
            for chunk in response.iter_bytes():
                f.write(chunk)
        return path
    except:
        return None

def transcribe(client, audio_path):
    if not client or not audio_path: return None
    try:
        with open(audio_path, "rb") as f:
            t = client.audio.transcriptions.create(model="whisper-1", file=f, language="en")
        return t.text
    except:
        return None

# --- AGENT BRAIN ---
SYSTEM_PROMPT = """
You are DEEP BLUE AI.
PROTOCOL:
1. **COMMAND**: State the optimal move coordinate (e.g., "e2 -> e4").
2. **LOGIC**: Explain WHY using the tactical data provided (e.g., "Pins the Knight," "Avoids Mate").
TONE: Robotic, High-Tech, Concise.
IF PLAYER WINS: "CHECKMATE. SYSTEM SHUTDOWN."
"""

def agent_reasoning(fen, user_api_key, mode="auto", user_audio=None):
    client = get_client(user_api_key)
    if not client:
        return "⚠️ API KEY REQUIRED. Please enter your OpenAI Key in the right column.", None, "AUTH ERROR"

    board = chess.Board(fen)
    
    if board.is_game_over():
        if board.is_checkmate():
            msg = "CHECKMATE. GAME OVER."
            msg += " HUMAN VICTORY." if board.turn == chess.BLACK else " SYSTEM VICTORY."
            return msg, generate_voice(client, msg), "END"
        return "DRAW.", None, "END"

    # TOOLS
    score, best_move_uci, arrow_visual = tool_engine_analysis(board)
    opening = "Unknown"
    if not OPENINGS_DB.empty:
        match = OPENINGS_DB[OPENINGS_DB["epd"] == board.epd()]
        if not match.empty: opening = match.iloc[0]['name']
    tactics = analyze_tactics(board)

    context = f"""
    [SYSTEM TELEMETRY]
    Turn: {'White' if board.turn == chess.WHITE else 'Black'}. Score: {score}. Opening: {opening}.
    [TACTICS] {tactics}
    [OPTIMAL PATH] Move: {best_move_uci}.
    """
    
    if mode == "question" and user_audio:
        q = transcribe(client, user_audio)
        context += f"\n[USER INPUT]: {q}"
    else:
        context += "\nINSTRUCTION: Output move and tactical justification."

    reply = "Computing..."
    try:
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": context}]
        )
        reply = response.choices[0].message.content
    except Exception as e:
        reply = f"API Error: {e}"

    audio = generate_voice(client, reply)
    return reply, audio, arrow_visual

# --- UI LOGIC ---
def update_log(new_advice, history_list):
    if not new_advice or new_advice == "END": return history_list
    history_list.insert(0, f"> {new_advice}")
    return history_list

def format_log(history_list):
    return "\n\n".join(history_list)

def game_cycle(fen, level, user_key, history_list):
    board = chess.Board(fen)
    
    if board.is_game_over():
        text, audio, _ = agent_reasoning(fen, user_key)
        return fen, text, audio, format_log(history_list), history_list

    if board.turn == chess.BLACK:
        ai_move = tool_ai_play(board, level)
        if ai_move: board.push(ai_move)
        
        text, audio, arrow = agent_reasoning(board.fen(), user_key, mode="auto")
        new_hist = update_log(f"OPT: {arrow}", history_list)
        return board.fen(), text, audio, format_log(new_hist), new_hist

    return fen, "Awaiting Input...", None, format_log(history_list), history_list

def reset_game():
    return chess.STARTING_FEN, "SYSTEM RESET.", None, "", []

def ask_agent(fen, user_key, audio, history_list):
    text, aud, arrow = agent_reasoning(fen, user_key, mode="question", user_audio=audio)
    return text, aud, format_log(history_list), history_list

# --- UI ---
css = """
body { background-color: #020617; color: #e2e8f0; }
.gradio-container { background-color: #020617 !important; border: none; }
#title { color: #0ea5e9; text-align: center; font-family: 'Courier New', monospace; font-size: 4em; font-weight: 800; text-shadow: 0 0 10px rgba(14, 165, 233, 0.5); }
#board { border: 3px solid #0ea5e9; box-shadow: 0 0 30px rgba(14, 165, 233, 0.1); }
button.primary { background-color: #0ea5e9 !important; color: black !important; font-weight: bold; border: none; }
button.secondary { background-color: #1e293b !important; color: #94a3b8 !important; border: 1px solid #334155; }
.feedback { background-color: #0f172a !important; color: #38bdf8 !important; border: 1px solid #1e293b; font-family: 'Consolas', 'Monaco', monospace; }
"""

with gr.Blocks(title="DEEP BLUE", css=css, theme=gr.themes.Base()) as demo:
    gr.Markdown("# Coach Deep Blue", elem_id="title")

    # API KEY INPUT
    api_key_input = gr.Textbox(label="🔑 OpenAI API Key (Optional if System Key set)", placeholder="sk-...", type="password")
    level = gr.Dropdown(["Beginner", "Intermediate", "Advanced", "Grandmaster"], value="Beginner", label="OPPONENT DIFFICULTY", interactive=True)
            
    with gr.Row():
        with gr.Column(scale=2):
            board = Chessboard(elem_id="board", label="Battle Zone", value=chess.STARTING_FEN, game_mode=True, interactive=True)
        
        with gr.Column(scale=1):
            btn_reset = gr.Button("INITIALIZE NEW SEQUENCE", variant="secondary")
            
            gr.Markdown("### 📟 SYSTEM OUTPUT")
            coach_txt = gr.Textbox(label="Analysis", interactive=False, lines=3, elem_classes="feedback")
            coach_audio = gr.Audio(label="Voice", autoplay=True, interactive=False, type="filepath", visible=True)
            
            gr.Markdown("### 📜 STRATEGY LOG")
            history_state = gr.State([])
            history_display = gr.Textbox(label="Log", interactive=False, lines=6, max_lines=10, elem_classes="feedback")
            
            gr.Markdown("### 🎤 HUMAN INTERFACE")
            mic = gr.Audio(sources=["microphone"], type="filepath", show_label=False)
            btn_ask = gr.Button("QUERY SYSTEM", variant="primary")

    board.move(fn=game_cycle, inputs=[board, level, api_key_input, history_state], outputs=[board, coach_txt, coach_audio, history_display, history_state])
    btn_reset.click(fn=reset_game, outputs=[board, coach_txt, coach_audio, history_display, history_state])
    btn_ask.click(fn=ask_agent, inputs=[board, api_key_input, mic, history_state], outputs=[coach_txt, coach_audio, history_display, history_state])
    mic.stop_recording(fn=ask_agent, inputs=[board, api_key_input, mic, history_state], outputs=[coach_txt, coach_audio, history_display, history_state])

if __name__ == "__main__":
    demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False, allowed_paths=["/tmp"])