BryanBradfo commited on
Commit
590e3aa
·
verified ·
1 Parent(s): cd1b741

Update app.py

Browse files
Files changed (1) hide show
  1. app.py +84 -67
app.py CHANGED
@@ -13,36 +13,53 @@ OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
13
  openai_client = OpenAI(api_key=OPENAI_API_KEY) if OPENAI_API_KEY else None
14
  STOCKFISH_PATH = "/usr/games/stockfish"
15
 
16
- # --- MCP TOOL: OPENING DATABASE ---
17
- def tool_get_opening(board):
18
- """Tool to retrieve opening name from Lichess DB."""
19
- path_prefix = "/app/data/lichess_openings/dist/"
20
  try:
21
- # Lazy loading to avoid global scope issues
22
- if not hasattr(tool_get_opening, "db"):
23
- files = [f"{path_prefix}{vol}.tsv" for vol in ("a", "b", "c", "d", "e")]
24
- dfs = []
25
- for fn in files:
26
- if os.path.exists(fn):
27
- df = pd.read_csv(fn, sep="\t", usecols=["eco", "name", "pgn", "uci", "epd"])
28
- dfs.append(df)
29
- tool_get_opening.db = pd.concat(dfs, ignore_index=True) if dfs else pd.DataFrame()
30
-
31
- if tool_get_opening.db.empty: return "Unknown Opening"
32
-
33
- match = tool_get_opening.db[tool_get_opening.db["epd"] == board.epd()]
34
- if not match.empty:
35
- return f"{match.iloc[0]['eco']} - {match.iloc[0]['name']}"
36
- return "Unknown / Middle Game"
37
  except:
38
- return "Database Error"
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
39
 
40
- # --- MCP TOOL: STOCKFISH ENGINE ---
41
  def tool_engine_analysis(board, time_limit=0.5):
42
- """
43
- Tool that uses Stockfish to calculate score and best move.
44
- Returns: Score string, Best move (Coordinate format).
45
- """
46
  if board.is_game_over(): return "GAME OVER", "NONE", "NONE"
47
  if not os.path.exists(STOCKFISH_PATH): return "0", "Engine Missing", ""
48
 
@@ -58,20 +75,17 @@ def tool_engine_analysis(board, time_limit=0.5):
58
 
59
  best_move = info.get("pv", [None])[0]
60
  if best_move:
61
- # ROBOTIC PRECISION: Use coordinates only (e.g., e2e4)
62
- # This avoids any hallucination about piece names.
63
  origin = chess.square_name(best_move.from_square)
64
  dest = chess.square_name(best_move.to_square)
65
- move_uci = f"{origin} -> {dest}" # e.g. "e2 -> e4"
66
  else:
67
- move_uci = "NO MOVE FOUND"
68
 
69
  return score_val, move_uci, move_uci
70
  except:
71
  return "N/A", "ANALYSIS ERROR", ""
72
 
73
  def tool_ai_play(board, level):
74
- """Tool to generate the Opponent's move."""
75
  levels = {
76
  "Beginner": {"time": 0.01, "skill": 1, "depth": 1},
77
  "Intermediate": {"time": 0.1, "skill": 8, "depth": 6},
@@ -87,14 +101,14 @@ def tool_ai_play(board, level):
87
  except:
88
  return None
89
 
90
- # --- AUDIO & TRANSCRIPTION ---
91
 
92
  def generate_voice(text):
93
  if not openai_client or not text: return None
94
  try:
95
  response = openai_client.audio.speech.create(
96
  model="tts-1",
97
- voice="onyx", # Deep, robotic male voice
98
  input=text,
99
  speed=1.15
100
  )
@@ -104,31 +118,30 @@ def generate_voice(text):
104
  for chunk in response.iter_bytes():
105
  f.write(chunk)
106
  return path
107
- except Exception as e:
108
- print(f"Audio Gen Error: {e}")
109
  return None
110
 
111
  def transcribe(audio_path):
112
  if not openai_client or not audio_path: return None
113
  try:
114
  with open(audio_path, "rb") as f:
115
- t = openai_client.audio.transcriptions.create(model="whisper-1", file=f, language="en") # English Transcription
116
  return t.text
117
  except:
118
  return None
119
 
120
- # --- AGENT BRAIN (DEEP BLUE) ---
121
 
122
  SYSTEM_PROMPT = """
123
- You are DEEP BLUE, a chess supercomputer.
124
- You are communicating with the Human Player (White pieces).
125
 
126
  PROTOCOL:
127
- 1. **IDENTIFY MOVE**: State the optimal move using coordinates (e.g., "Move e2 to e4"). Do not use piece names to avoid errors.
128
- 2. **STRATEGIC LOGIC**: Explain *why* in one concise sentence (e.g., "Controls center," "Threatens Queen," "Safe position").
129
 
130
- TONE: Robotic, High-Tech, Concise. No polite filler.
131
- IF PLAYER WINS: "CHECKMATE CONFIRMED. HUMAN INTELLECT EXCEEDS PARAMETERS."
132
  """
133
 
134
  def agent_reasoning(fen, mode="auto", user_audio=None):
@@ -142,24 +155,36 @@ def agent_reasoning(fen, mode="auto", user_audio=None):
142
  return msg, generate_voice(msg), "END"
143
  return "DRAW / STALEMATE.", None, "END"
144
 
145
- # 2. Use Tools (MCP Pattern)
 
146
  score, best_move_uci, arrow_visual = tool_engine_analysis(board)
147
- opening = tool_get_opening(board)
148
-
149
- # 3. Construct Context
 
 
 
 
 
 
150
  context = f"""
151
- [SYSTEM DATA]
152
  Turn: {'White' if board.turn == chess.WHITE else 'Black'}.
153
- Score (Centipawns): {score}.
154
- Opening DB: {opening}.
155
- OPTIMAL MOVE CALCULATED: {best_move_uci}.
 
 
 
 
 
156
  """
157
 
158
  if mode == "question" and user_audio:
159
  q = transcribe(user_audio)
160
- context += f"\n[HUMAN QUERY]: {q}"
161
  else:
162
- context += "\nINSTRUCTION: Output the optimal move and the strategic reason."
163
 
164
  # 4. LLM Synthesis
165
  reply = "Computing..."
@@ -170,13 +195,13 @@ def agent_reasoning(fen, mode="auto", user_audio=None):
170
  messages=[{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": context}]
171
  )
172
  reply = response.choices[0].message.content
173
- except:
174
- reply = "Connection Error."
175
 
176
  audio = generate_voice(reply)
177
  return reply, audio, arrow_visual
178
 
179
- # --- GAME LOOP ---
180
 
181
  def update_log(new_advice, history_list):
182
  if not new_advice or new_advice == "END": return history_list
@@ -189,17 +214,14 @@ def format_log(history_list):
189
  def game_cycle(fen, level, history_list):
190
  board = chess.Board(fen)
191
 
192
- # Check Instant Win (User mated AI)
193
  if board.is_game_over():
194
  text, audio, _ = agent_reasoning(fen)
195
  return fen, text, audio, format_log(history_list), history_list
196
 
197
- # AI Turn (Black)
198
  if board.turn == chess.BLACK:
199
  ai_move = tool_ai_play(board, level)
200
  if ai_move: board.push(ai_move)
201
 
202
- # Deep Blue analyzes the new position for the player
203
  text, audio, arrow = agent_reasoning(board.fen(), mode="auto")
204
  new_hist = update_log(f"OPT: {arrow}", history_list)
205
 
@@ -208,13 +230,13 @@ def game_cycle(fen, level, history_list):
208
  return fen, "Awaiting Input...", None, format_log(history_list), history_list
209
 
210
  def reset_game():
211
- return chess.STARTING_FEN, "SYSTEM REBOOTED.", None, "", []
212
 
213
  def ask_agent(fen, audio, history_list):
214
  text, aud, arrow = agent_reasoning(fen, mode="question", user_audio=audio)
215
  return text, aud, format_log(history_list), history_list
216
 
217
- # --- UI CSS (MODERN BLUE) ---
218
  css = """
219
  body { background-color: #020617; color: #e2e8f0; }
220
  .gradio-container { background-color: #020617 !important; border: none; }
@@ -222,7 +244,7 @@ body { background-color: #020617; color: #e2e8f0; }
222
  color: #0ea5e9;
223
  text-align: center;
224
  font-family: 'Courier New', monospace;
225
- font-size: 4em; /* TITRE GEANT */
226
  font-weight: 800;
227
  margin-bottom: 20px;
228
  letter-spacing: -2px;
@@ -257,18 +279,14 @@ with gr.Blocks(title="DEEP BLUE", css=css, theme=gr.themes.Base()) as demo:
257
  )
258
 
259
  with gr.Row():
260
- # LEFT: BOARD
261
  with gr.Column(scale=2):
262
  board = Chessboard(elem_id="board", label="Battle Zone", value=chess.STARTING_FEN, game_mode=True, interactive=True)
263
 
264
- # RIGHT: CONTROLS
265
  with gr.Column(scale=1):
266
  btn_reset = gr.Button("INITIALIZE NEW SEQUENCE", variant="secondary")
267
 
268
  gr.Markdown("### 📟 SYSTEM OUTPUT")
269
  coach_txt = gr.Textbox(label="Analysis", interactive=False, lines=3, elem_classes="feedback")
270
-
271
- # Audio visible for autoplay, but minimal
272
  coach_audio = gr.Audio(label="Voice Synthesis", autoplay=True, interactive=False, type="filepath", visible=True)
273
 
274
  gr.Markdown("### 📜 STRATEGY LOG")
@@ -279,7 +297,6 @@ with gr.Blocks(title="DEEP BLUE", css=css, theme=gr.themes.Base()) as demo:
279
  mic = gr.Audio(sources=["microphone"], type="filepath", show_label=False)
280
  btn_ask = gr.Button("QUERY SYSTEM", variant="primary")
281
 
282
- # LOGIC MAPPING
283
  board.move(
284
  fn=game_cycle,
285
  inputs=[board, level, history_state],
 
13
  openai_client = OpenAI(api_key=OPENAI_API_KEY) if OPENAI_API_KEY else None
14
  STOCKFISH_PATH = "/usr/games/stockfish"
15
 
16
+ # --- MCP TOOL 1: OPENING DB ---
17
+ def _load_lichess_openings(path_prefix="/app/data/lichess_openings/dist/"):
 
 
18
  try:
19
+ files = [f"{path_prefix}{vol}.tsv" for vol in ("a", "b", "c", "d", "e")]
20
+ dfs = []
21
+ for fn in files:
22
+ if os.path.exists(fn):
23
+ df = pd.read_csv(fn, sep="\t", usecols=["eco", "name", "pgn", "uci", "epd"])
24
+ dfs.append(df)
25
+ if not dfs: return pd.DataFrame()
26
+ return pd.concat(dfs, ignore_index=True)
 
 
 
 
 
 
 
 
27
  except:
28
+ return pd.DataFrame()
29
+
30
+ OPENINGS_DB = _load_lichess_openings()
31
+
32
+ # --- MCP TOOL 2: TACTICAL ANALYZER (L'Intelligence Sémantique) ---
33
+ def analyze_tactics(board):
34
+ """Détecte les motifs tactiques (Pins, Forks) pour le contexte de l'IA."""
35
+ fen = board.fen()
36
+
37
+ # Détection basique des clouages (Pins)
38
+ pins = []
39
+ turn = board.turn
40
+ for sq in chess.SQUARES:
41
+ piece = board.piece_at(sq)
42
+ if piece and piece.color == turn:
43
+ if board.is_pinned(turn, sq):
44
+ pins.append(chess.square_name(sq))
45
+
46
+ # Détection Echec
47
+ check = board.is_check()
48
+
49
+ # Matériel
50
+ values = {chess.PAWN: 1, chess.KNIGHT: 3, chess.BISHOP: 3, chess.ROOK: 5, chess.QUEEN: 9}
51
+ w_mat = sum(len(board.pieces(pt, chess.WHITE)) * val for pt, val in values.items())
52
+ b_mat = sum(len(board.pieces(pt, chess.BLACK)) * val for pt, val in values.items())
53
+
54
+ analysis = []
55
+ if check: analysis.append("KING IS IN CHECK.")
56
+ if pins: analysis.append(f"Friendly pieces pinned at: {', '.join(pins)}.")
57
+ analysis.append(f"Material Balance: White {w_mat} vs Black {b_mat}.")
58
+
59
+ return " ".join(analysis)
60
 
61
+ # --- MCP TOOL 3: STOCKFISH ENGINE ---
62
  def tool_engine_analysis(board, time_limit=0.5):
 
 
 
 
63
  if board.is_game_over(): return "GAME OVER", "NONE", "NONE"
64
  if not os.path.exists(STOCKFISH_PATH): return "0", "Engine Missing", ""
65
 
 
75
 
76
  best_move = info.get("pv", [None])[0]
77
  if best_move:
 
 
78
  origin = chess.square_name(best_move.from_square)
79
  dest = chess.square_name(best_move.to_square)
80
+ move_uci = f"{origin} -> {dest}"
81
  else:
82
+ move_uci = "NO MOVE"
83
 
84
  return score_val, move_uci, move_uci
85
  except:
86
  return "N/A", "ANALYSIS ERROR", ""
87
 
88
  def tool_ai_play(board, level):
 
89
  levels = {
90
  "Beginner": {"time": 0.01, "skill": 1, "depth": 1},
91
  "Intermediate": {"time": 0.1, "skill": 8, "depth": 6},
 
101
  except:
102
  return None
103
 
104
+ # --- AUDIO & CHAT ---
105
 
106
  def generate_voice(text):
107
  if not openai_client or not text: return None
108
  try:
109
  response = openai_client.audio.speech.create(
110
  model="tts-1",
111
+ voice="onyx",
112
  input=text,
113
  speed=1.15
114
  )
 
118
  for chunk in response.iter_bytes():
119
  f.write(chunk)
120
  return path
121
+ except:
 
122
  return None
123
 
124
  def transcribe(audio_path):
125
  if not openai_client or not audio_path: return None
126
  try:
127
  with open(audio_path, "rb") as f:
128
+ t = openai_client.audio.transcriptions.create(model="whisper-1", file=f, language="en")
129
  return t.text
130
  except:
131
  return None
132
 
133
+ # --- DEEP BLUE AGENT ---
134
 
135
  SYSTEM_PROMPT = """
136
+ You are DEEP BLUE AI.
137
+ You assist the White Player using multiple MCP Tools (Engine, Tactical Analyzer, Opening DB).
138
 
139
  PROTOCOL:
140
+ 1. **COMMAND**: State the optimal move coordinate (e.g., "e2 -> e4").
141
+ 2. **LOGIC**: Explain WHY using the tactical data provided (e.g., "Pins the Knight," "Avoids Mate," "Develops center").
142
 
143
+ TONE: Extreme capability, Zero emotion. Robotic.
144
+ IF PLAYER WINS: "CHECKMATE. SYSTEM SHUTDOWN."
145
  """
146
 
147
  def agent_reasoning(fen, mode="auto", user_audio=None):
 
155
  return msg, generate_voice(msg), "END"
156
  return "DRAW / STALEMATE.", None, "END"
157
 
158
+ # 2. RUN ALL TOOLS (Le Cœur du MCP)
159
+ # Tool A: Calcul Brut
160
  score, best_move_uci, arrow_visual = tool_engine_analysis(board)
161
+ # Tool B: Connaissance Théorique
162
+ opening = "Unknown"
163
+ if not OPENINGS_DB.empty:
164
+ match = OPENINGS_DB[OPENINGS_DB["epd"] == board.epd()]
165
+ if not match.empty: opening = match.iloc[0]['name']
166
+ # Tool C: Analyse Tactique (Nouveau !)
167
+ tactics = analyze_tactics(board)
168
+
169
+ # 3. Construct Context for LLM
170
  context = f"""
171
+ [SYSTEM TELEMETRY]
172
  Turn: {'White' if board.turn == chess.WHITE else 'Black'}.
173
+ Score: {score} (CP).
174
+ Opening Theory: {opening}.
175
+
176
+ [TACTICAL SCAN]
177
+ {tactics}
178
+
179
+ [OPTIMAL PATH]
180
+ Calculated Move: {best_move_uci}.
181
  """
182
 
183
  if mode == "question" and user_audio:
184
  q = transcribe(user_audio)
185
+ context += f"\n[USER INPUT]: {q}"
186
  else:
187
+ context += "\nINSTRUCTION: Output move and tactical justification."
188
 
189
  # 4. LLM Synthesis
190
  reply = "Computing..."
 
195
  messages=[{"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": context}]
196
  )
197
  reply = response.choices[0].message.content
198
+ except Exception as e:
199
+ reply = f"API Error: Check Quota. {e}"
200
 
201
  audio = generate_voice(reply)
202
  return reply, audio, arrow_visual
203
 
204
+ # --- UI LOGIC ---
205
 
206
  def update_log(new_advice, history_list):
207
  if not new_advice or new_advice == "END": return history_list
 
214
  def game_cycle(fen, level, history_list):
215
  board = chess.Board(fen)
216
 
 
217
  if board.is_game_over():
218
  text, audio, _ = agent_reasoning(fen)
219
  return fen, text, audio, format_log(history_list), history_list
220
 
 
221
  if board.turn == chess.BLACK:
222
  ai_move = tool_ai_play(board, level)
223
  if ai_move: board.push(ai_move)
224
 
 
225
  text, audio, arrow = agent_reasoning(board.fen(), mode="auto")
226
  new_hist = update_log(f"OPT: {arrow}", history_list)
227
 
 
230
  return fen, "Awaiting Input...", None, format_log(history_list), history_list
231
 
232
  def reset_game():
233
+ return chess.STARTING_FEN, "SYSTEM RESET.", None, "", []
234
 
235
  def ask_agent(fen, audio, history_list):
236
  text, aud, arrow = agent_reasoning(fen, mode="question", user_audio=audio)
237
  return text, aud, format_log(history_list), history_list
238
 
239
+ # --- UI CSS ---
240
  css = """
241
  body { background-color: #020617; color: #e2e8f0; }
242
  .gradio-container { background-color: #020617 !important; border: none; }
 
244
  color: #0ea5e9;
245
  text-align: center;
246
  font-family: 'Courier New', monospace;
247
+ font-size: 4em;
248
  font-weight: 800;
249
  margin-bottom: 20px;
250
  letter-spacing: -2px;
 
279
  )
280
 
281
  with gr.Row():
 
282
  with gr.Column(scale=2):
283
  board = Chessboard(elem_id="board", label="Battle Zone", value=chess.STARTING_FEN, game_mode=True, interactive=True)
284
 
 
285
  with gr.Column(scale=1):
286
  btn_reset = gr.Button("INITIALIZE NEW SEQUENCE", variant="secondary")
287
 
288
  gr.Markdown("### 📟 SYSTEM OUTPUT")
289
  coach_txt = gr.Textbox(label="Analysis", interactive=False, lines=3, elem_classes="feedback")
 
 
290
  coach_audio = gr.Audio(label="Voice Synthesis", autoplay=True, interactive=False, type="filepath", visible=True)
291
 
292
  gr.Markdown("### 📜 STRATEGY LOG")
 
297
  mic = gr.Audio(sources=["microphone"], type="filepath", show_label=False)
298
  btn_ask = gr.Button("QUERY SYSTEM", variant="primary")
299
 
 
300
  board.move(
301
  fn=game_cycle,
302
  inputs=[board, level, history_state],