doctorlinux commited on
Commit
4c71f30
·
verified ·
1 Parent(s): 4ea4ead

Delete app.py

Browse files
Files changed (1) hide show
  1. app.py +0 -342
app.py DELETED
@@ -1,342 +0,0 @@
1
- import os
2
- import io
3
- import re
4
- import math
5
- import traceback
6
- from io import StringIO
7
- import gradio as gr
8
- import chess, chess.pgn, chess.engine
9
- import numpy as np
10
- import pandas as pd
11
- import matplotlib
12
- matplotlib.use("Agg")
13
- import matplotlib.pyplot as plt
14
-
15
- APP_TITLE = "AI Chess Analyzer — estilo DecodeChess (Doctor Linux)"
16
- ENGINE_PATHS = ["stockfish", "/usr/bin/stockfish", "/usr/games/stockfish"]
17
-
18
- # -------------------------------
19
- # Carga opcional de modelo ML (blunders)
20
- # -------------------------------
21
- blunder_model = None
22
- blunder_features = None
23
- try:
24
- from joblib import load as joblib_load
25
- f_model = "models/blunder/model.joblib"
26
- f_feats = "models/blunder/features.txt"
27
- if os.path.exists(f_model) and os.path.exists(f_feats):
28
- blunder_model = joblib_load(f_model)
29
- with open(f_feats, "r", encoding="utf-8") as f:
30
- blunder_features = [ln.strip() for ln in f if ln.strip()]
31
- print("✅ Modelo de blunders cargado.")
32
- except Exception as e:
33
- print("⚠️ No se pudo cargar el modelo de blunders:", e)
34
-
35
- # -------------------------------
36
- # Reparador PGN
37
- # -------------------------------
38
- class ReparadorPGN:
39
- @staticmethod
40
- def reparar_pgn(pgn_text: str) -> str:
41
- if not isinstance(pgn_text, str):
42
- return pgn_text
43
- lineas = pgn_text.splitlines()
44
- out = []
45
- for linea in lineas:
46
- original = linea
47
- s = linea.strip()
48
- if s.startswith("[") and "]" in s:
49
- s = re.sub(r'\[([A-Za-z0-9_]+)\s+"([^"]*)["“”]?\]?$', r'[\1 "\2"]', s)
50
- s = re.sub(r'\[Ulnite', '[White', s)
51
- s = s.replace('[Result "I-0"]', '[Result "1-0"]')
52
- s = s.replace('[Result "O-I"]', '[Result "0-1"]')
53
- s = s.replace('[Result "I/2-I/2"]', '[Result "1/2-1/2"]')
54
- out.append(s); continue
55
- t = s
56
- correcciones = {
57
- r'\bnf([1-8a-h])': r'Nf\1',
58
- r'\bnc([1-8a-h])': r'Nc\1',
59
- r'\bng([1-8a-h])': r'Ng\1',
60
- r'\bhc([1-8])\b': 'Nc\\1',
61
- r'\bhf([1-8])\b': 'Nf\\1',
62
- r'\bnn1\b': 'Nf1',
63
- r'\bhe2\b': 'Ne2',
64
- r'\bnh7\b': 'Nh7',
65
- r'\bhc5\b': 'Nc5',
66
- r'\bqu2\b': 'Qd2',
67
- r'\bre1\b': 'Re1',
68
- r'\brn\b': 'Rf1',
69
- r'\bbe4:?': 'Be4',
70
- r'\bo-o-o\b': 'O-O-O',
71
- r'\bo-o\b': 'O-O',
72
- }
73
- for pat, rep in correcciones.items():
74
- t = re.sub(pat, rep, t, flags=re.IGNORECASE)
75
- out.append(t if t else original)
76
- texto = "\n".join(out)
77
- texto = texto.replace('Result "* *"', 'Result "*"')
78
- return texto
79
-
80
- # -------------------------------
81
- # Motor de ajedrez
82
- # -------------------------------
83
- def load_engine():
84
- last_err = None
85
- for p in ENGINE_PATHS:
86
- try:
87
- eng = chess.engine.SimpleEngine.popen_uci(p)
88
- return eng
89
- except Exception as e:
90
- last_err = e
91
- raise RuntimeError(f"No pude iniciar Stockfish. ¿Está instalado? Último error: {last_err}")
92
-
93
- def score_to_cp(score: chess.engine.PovScore) -> float:
94
- # Devuelve evaluación en centipawns desde el punto de vista del bando que juega
95
- if score.is_mate():
96
- # usar un valor grande con signo para graficar
97
- mate = score.white().mate()
98
- if mate is None:
99
- return 0.0
100
- return 100000.0 if mate > 0 else -100000.0
101
- return float(score.white().score(mate_score=100000))
102
-
103
- def classify_drop(delta_cp: float, mate_change: int|None) -> str:
104
- # delta_cp = eval_after - eval_before (POV del bando que juega antes de mover)
105
- # Si delta es muy negativo => caída de evaluación => peor jugada
106
- if mate_change is not None and mate_change < 0:
107
- return "Blunder (perdió/permitió mate)"
108
- drop = -delta_cp
109
- if drop < 20: return "Best/Excellent"
110
- if drop < 60: return "Good"
111
- if drop < 120: return "Inaccuracy"
112
- if drop < 300: return "Mistake"
113
- return "Blunder"
114
-
115
- def natural_explanation(delta_cp, best_san, played_san, b_before: chess.Board, b_after: chess.Board, info_before, info_after) -> str:
116
- tips = []
117
- if info_before.get("score") and info_after.get("score"):
118
- sb = info_before["score"]; sa = info_after["score"]
119
- if sb.is_mate() and not sa.is_mate():
120
- tips.append("Se perdió una secuencia de mate forzado.")
121
- if not sb.is_mate() and sa.is_mate():
122
- tips.append("Se permitió una secuencia de mate forzado.")
123
- if -delta_cp >= 300:
124
- tips.append("La evaluación cayó fuertemente; revisa táctica inmediata (piezas colgando, mates).")
125
- elif -delta_cp >= 120:
126
- tips.append("Cede ventaja significativa; había opciones más fuertes.")
127
- elif -delta_cp >= 60:
128
- tips.append("Había una alternativa mejor según el motor.")
129
- if b_after.is_check():
130
- tips.append("La jugada conduce a jaques del rival o deja al rey expuesto.")
131
- center = [chess.D4, chess.E4, chess.D5, chess.E5]
132
- if any(b_after.piece_at(sq) and b_after.piece_at(sq).piece_type==chess.PAWN for sq in center):
133
- tips.append("Buen control del centro con peones.")
134
- if best_san and played_san and best_san != played_san:
135
- tips.append(f"Recomendación del motor: {best_san} en lugar de {played_san}.")
136
- if not tips:
137
- tips.append("Jugada razonable.")
138
- return " ".join(tips)
139
-
140
- def analyze_game_with_engine(game: chess.pgn.Game, engine, time_limit=0.3, depth=None):
141
- board = game.board()
142
- ann_rows = []
143
- eval_cp_series = []
144
- annotated_pgn = io.StringIO()
145
-
146
- # Exporter para re-serializar con comentarios
147
- exporter = chess.pgn.StringExporter(headers=True, variations=False, comments=True)
148
-
149
- node = game
150
- move_index = 0
151
- while node.variations:
152
- move = node.variation(0).move
153
- turn_white = board.turn # bando que mueve antes de la jugada
154
- # eval antes
155
- info_before = engine.analyse(board, chess.engine.Limit(time=time_limit, depth=depth))
156
- eval_before = score_to_cp(info_before["score"].pov(chess.WHITE))
157
- # mejor jugada sugerida
158
- best_move = info_before.get("pv", [move])[0]
159
- best_san = board.san(best_move) if best_move else None
160
-
161
- # jugar jugada real
162
- played_san = board.san(move)
163
- board.push(move)
164
-
165
- # eval después
166
- info_after = engine.analyse(board, chess.engine.Limit(time=time_limit, depth=depth))
167
- eval_after = score_to_cp(info_after["score"].pov(chess.WHITE))
168
-
169
- # delta desde la perspectiva del bando que jugaba
170
- delta_cp = (eval_after if turn_white else -eval_after) - (eval_before if turn_white else -eval_before)
171
- mate_change = None
172
- if info_before["score"].is_mate() or info_after["score"].is_mate():
173
- # si empeora la distancia a mate desde POV del bando que jugaba, marcamos negativo
174
- m_before = info_before["score"].white().mate()
175
- m_after = info_after["score"].white().mate()
176
- if m_before is not None and m_after is not None:
177
- mate_change = abs(m_before) - abs(m_after)
178
-
179
- category = classify_drop(delta_cp, mate_change)
180
- explanation = natural_explanation(delta_cp, best_san, played_san, node.board(), board, info_before, info_after)
181
-
182
- eval_cp_series.append(eval_after if board.turn else -eval_after)
183
-
184
- # guardar fila
185
- move_index += 1
186
- ann_rows.append({
187
- "ply": move_index,
188
- "turn": "White" if turn_white else "Black",
189
- "played": played_san,
190
- "best": best_san,
191
- "delta_cp": round(delta_cp, 1),
192
- "eval_after_cp": round(eval_after, 1),
193
- "category": category,
194
- "explanation": explanation
195
- })
196
-
197
- # comentario en el nodo siguiente
198
- node = node.variation(0)
199
- if node.comment:
200
- node.comment += " "
201
- else:
202
- node.comment = ""
203
- node.comment += f"[{category}] Δ={round(delta_cp,1)} | Mejor: {best_san}. {explanation}"
204
-
205
- annotated_text = game.accept(exporter)
206
- return ann_rows, eval_cp_series, annotated_text
207
-
208
- # -------------------------------
209
- # Gráfica
210
- # -------------------------------
211
- def plot_eval(pgn_headers, eval_series):
212
- fig, ax = plt.subplots(figsize=(12, 5))
213
- if not eval_series:
214
- ax.text(0.5,0.5,"Sin evaluación (¿PGN vacío?)", ha="center", va="center", transform=ax.transAxes)
215
- ax.set_axis_off()
216
- return fig
217
- xs = list(range(1, len(eval_series)+1))
218
- ax.plot(xs, eval_series, linewidth=2)
219
- ax.axhline(0, linestyle="--")
220
- ax.set_xlabel("Ply")
221
- ax.set_ylabel("Evaluación (cp, + = Blancas)")
222
- title = f"{pgn_headers.get('White','?')} vs {pgn_headers.get('Black','?')} — {pgn_headers.get('Result','*')}"
223
- ax.set_title(title)
224
- ax.grid(True, alpha=0.3)
225
- fig.tight_layout()
226
- return fig
227
-
228
- # -------------------------------
229
- # Pipeline principal
230
- # -------------------------------
231
- def process(pgn_text, time_per_move, depth_limit):
232
- if not pgn_text or not pgn_text.strip():
233
- return None, "Pega un PGN.", None, None, None
234
-
235
- repaired = ReparadorPGN.reparar_pgn(pgn_text)
236
-
237
- # Leer primera partida válida
238
- f = StringIO(repaired)
239
- game = chess.pgn.read_game(f)
240
- while game is not None and sum(1 for _ in game.mainline_moves()) == 0:
241
- game = chess.pgn.read_game(f)
242
-
243
- if game is None:
244
- return None, "No se encontró una partida válida.", repaired, None, None
245
-
246
- # Motor
247
- try:
248
- engine = load_engine()
249
- except Exception as e:
250
- err = f"No pude iniciar Stockfish: {e}"
251
- return None, err, repaired, None, None
252
-
253
- try:
254
- rows, evals, annotated_pgn = analyze_game_with_engine(game, engine, time_limit=time_per_move, depth=None if depth_limit<=0 else depth_limit)
255
- except Exception as e:
256
- engine.quit()
257
- tb = traceback.format_exc(limit=2)
258
- return None, f"Falló el análisis: {e}\n{tb}", repaired, None, None
259
-
260
- engine.quit()
261
-
262
- # Si hay modelo ML, añadimos prob. blunder
263
- if blunder_model is not None and blunder_features is not None and rows:
264
- df = pd.DataFrame(rows)
265
- # construir features simples por ahora
266
- df_feat = df.copy()
267
- for col in blunder_features:
268
- if col not in df_feat.columns:
269
- df_feat[col] = 0.0
270
- try:
271
- proba = blunder_model.predict_proba(df_feat[blunder_features].astype(float).values)[:,1]
272
- df["blunder_proba"] = np.round(proba, 3)
273
- rows = df.to_dict(orient="records")
274
- except Exception as e:
275
- print("⚠️ No se pudo inferir prob. blunder:", e)
276
-
277
- # Render tabla markdown resumen top errores
278
- worst = sorted(rows, key=lambda r: r["delta_cp"])[:10] # más caída (delta más negativo)
279
- md_lines = ["### Resumen (top caídas de evaluación)",
280
- "| Ply | Turno | Jugada | Mejor | Δcp | Categoría |",
281
- "|---:|:---:|:---|:---|---:|:---|"]
282
- for r in worst:
283
- md_lines.append(f"| {r['ply']} | {r['turn']} | {r['played']} | {r.get('best','')} | {r['delta_cp']} | {r['category']} |")
284
- md_report = "\n".join(md_lines)
285
-
286
- fig = plot_eval(game.headers, evals)
287
-
288
- # CSV para descargar
289
- import csv
290
- csv_buf = io.StringIO()
291
- cw = csv.DictWriter(csv_buf, fieldnames=list(rows[0].keys()))
292
- cw.writeheader()
293
- cw.writerows(rows)
294
- csv_bytes = csv_buf.getvalue()
295
-
296
- return fig, md_report, repaired, annotated_pgn, csv_bytes
297
-
298
- # -------------------------------
299
- # UI Gradio
300
- # -------------------------------
301
- with gr.Blocks(title=APP_TITLE) as demo:
302
- gr.Markdown(f"# {APP_TITLE}\nCarga un PGN. Se analiza la **primera** partida válida.\n- Motor: Stockfish (apt)\n- Clasificación: Best/Good/Inaccuracy/Mistake/Blunder\n- Comentarios automáticos estilo DecodeChess (explicación en lenguaje natural)\n- Descarga PGN anotado + CSV por jugada")
303
-
304
- with gr.Row():
305
- pgn_in = gr.Textbox(lines=18, label="PGN (pegar aquí)", placeholder="Pega aquí el PGN...")
306
-
307
- with gr.Row():
308
- time_per = gr.Slider(0.05, 1.0, value=0.25, step=0.05, label="Tiempo por jugada (s)")
309
- depth = gr.Slider(0, 30, value=0, step=1, label="Límite de profundidad (0 = solo tiempo)")
310
-
311
- run_btn = gr.Button("Analizar")
312
-
313
- with gr.Row():
314
- plot_out = gr.Plot(label="Gráfico de evaluación")
315
- with gr.Row():
316
- md_out = gr.Markdown(label="Resumen")
317
- with gr.Row():
318
- repaired_out = gr.Textbox(lines=10, label="PGN reparado (no sobrescribe tu original)")
319
-
320
- with gr.Row():
321
- ann_pgn = gr.Textbox(lines=12, label="PGN anotado (descargable)")
322
- with gr.Row():
323
- dl_pgn = gr.File(label="Descargar PGN anotado")
324
- dl_csv = gr.File(label="Descargar CSV jugadas")
325
-
326
- def _run_and_pack(pgn_text, time_per_move, depth_limit):
327
- fig, md, repaired, pgn_annot, csv_bytes = process(pgn_text, time_per_move, depth_limit)
328
- files = []
329
- if pgn_annot:
330
- fn = "annotated.pgn"
331
- open(fn, "w", encoding="utf-8").write(pgn_annot)
332
- files.append(fn)
333
- if csv_bytes:
334
- fn2 = "moves.csv"
335
- open(fn2, "w", encoding="utf-8").write(csv_bytes)
336
- files.append(fn2)
337
- return fig, md, repaired, pgn_annot, files[0] if files else None, files[1] if len(files)>1 else None
338
-
339
- run_btn.click(_run_and_pack, inputs=[pgn_in, time_per, depth], outputs=[plot_out, md_out, repaired_out, ann_pgn, dl_pgn, dl_csv])
340
-
341
- if __name__ == "__main__":
342
- demo.launch()