""" CASCADE-LATTICE Chess Demo Clean rebuild with Dash + Three.js - Human plays White (select from HOLD candidates, then COMMIT) - Engine plays Black (auto-responds after COMMIT) - Full informational wealth display (features, reasoning, imagination) """ import os import sys import platform import asyncio import shutil import threading import time from pathlib import Path from dataclasses import dataclass from typing import List, Dict, Any, Optional import chess import chess.engine import numpy as np import dash from dash import dcc, html, callback, Input, Output, State, ctx import json # ═══════════════════════════════════════════════════════════════════════════════ # ENGINE SETUP # ═══════════════════════════════════════════════════════════════════════════════ PROJECT_ROOT = Path(__file__).parent.parent LOCAL_STOCKFISH_SSE41 = PROJECT_ROOT / "stockfish" / "stockfish" / "stockfish-windows-x86-64-sse41-popcnt.exe" LOCAL_STOCKFISH_AVX2 = PROJECT_ROOT / "stockfish" / "stockfish-windows-x86-64-avx2.exe" if LOCAL_STOCKFISH_SSE41.exists(): STOCKFISH_PATH = str(LOCAL_STOCKFISH_SSE41) elif LOCAL_STOCKFISH_AVX2.exists(): STOCKFISH_PATH = str(LOCAL_STOCKFISH_AVX2) elif shutil.which("stockfish"): STOCKFISH_PATH = shutil.which("stockfish") else: STOCKFISH_PATH = "/usr/games/stockfish" print(f"[STOCKFISH] {STOCKFISH_PATH}") if platform.system() == 'Windows': asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy()) # Thread-safe engine ENGINE_LOCK = threading.Lock() ENGINE = None def get_engine(): global ENGINE with ENGINE_LOCK: if ENGINE is None: try: ENGINE = chess.engine.SimpleEngine.popen_uci(STOCKFISH_PATH) print("[ENGINE] Started") except Exception as e: print(f"[ENGINE] Failed: {e}") return ENGINE def engine_analyse(board, limit, multipv=1): """Thread-safe engine analysis.""" global ENGINE with ENGINE_LOCK: if ENGINE is None: get_engine() if ENGINE: try: return ENGINE.analyse(board, limit, multipv=multipv) except chess.engine.EngineTerminatedError: print("[ENGINE] Crashed, restarting...") ENGINE = None get_engine() if ENGINE: return ENGINE.analyse(board, limit, multipv=multipv) return [] def engine_play(board, limit): """Thread-safe engine play.""" global ENGINE with ENGINE_LOCK: if ENGINE is None: get_engine() if ENGINE: try: return ENGINE.play(board, limit) except chess.engine.EngineTerminatedError: print("[ENGINE] Crashed, restarting...") ENGINE = None get_engine() if ENGINE: return ENGINE.play(board, limit) return None # Initial engine load get_engine() # ═══════════════════════════════════════════════════════════════════════════════ # CASCADE-LATTICE # ═══════════════════════════════════════════════════════════════════════════════ CASCADE_AVAILABLE = False HOLD = None try: from cascade import Hold HOLD = Hold() CASCADE_AVAILABLE = True print("[CASCADE] Hold ready") except ImportError: print("[CASCADE] Hold not available") try: from cascade import CausationGraph CAUSATION = CausationGraph() print("[CASCADE] CausationGraph ready") except: CAUSATION = None # ═══════════════════════════════════════════════════════════════════════════════ # DATA STRUCTURES # ═══════════════════════════════════════════════════════════════════════════════ @dataclass class MoveCandidate: move: str prob: float value: float from_sq: int to_sq: int is_capture: bool is_check: bool move_type: str # 'quiet', 'capture', 'check', 'castle' # ═══════════════════════════════════════════════════════════════════════════════ # FEATURE EXTRACTION # ═══════════════════════════════════════════════════════════════════════════════ PIECE_VALUES = {chess.PAWN: 1, chess.KNIGHT: 3, chess.BISHOP: 3, chess.ROOK: 5, chess.QUEEN: 9, chess.KING: 0} def extract_features(board: chess.Board) -> Dict[str, Any]: """Extract position features for cascade display.""" # Material white_mat = sum(len(board.pieces(pt, chess.WHITE)) * PIECE_VALUES[pt] for pt in PIECE_VALUES) black_mat = sum(len(board.pieces(pt, chess.BLACK)) * PIECE_VALUES[pt] for pt in PIECE_VALUES) material = white_mat - black_mat # Center control center = [chess.D4, chess.D5, chess.E4, chess.E5] white_center = sum(1 for sq in center if board.is_attacked_by(chess.WHITE, sq)) black_center = sum(1 for sq in center if board.is_attacked_by(chess.BLACK, sq)) # King safety wk = board.king(chess.WHITE) bk = board.king(chess.BLACK) white_king_attackers = len(board.attackers(chess.BLACK, wk)) if wk else 0 black_king_attackers = len(board.attackers(chess.WHITE, bk)) if bk else 0 # Mobility turn = board.turn board.turn = chess.WHITE white_mobility = len(list(board.legal_moves)) board.turn = chess.BLACK black_mobility = len(list(board.legal_moves)) board.turn = turn # Game phase total_pieces = len(board.piece_map()) if total_pieces > 28: phase = 'opening' elif total_pieces > 14: phase = 'middlegame' else: phase = 'endgame' return { 'material': material, 'center_control': (white_center - black_center) / 4.0, 'white_king_danger': white_king_attackers, 'black_king_danger': black_king_attackers, 'white_mobility': white_mobility, 'black_mobility': black_mobility, 'phase': phase } # ═══════════════════════════════════════════════════════════════════════════════ # CANDIDATE GENERATION WITH CASCADE INTEGRATION # ═══════════════════════════════════════════════════════════════════════════════ def get_candidates_with_hold(board: chess.Board, num=5) -> tuple: """ Get move candidates from engine + build cascade Hold data. Returns: (candidates, hold_data) """ candidates = [] hold_data = {} info = engine_analyse(board, chess.engine.Limit(depth=12), multipv=num) if not info: return [], {} features = extract_features(board) action_labels = [] raw_values = [] for pv in info: move = pv['pv'][0] score = pv.get('score', chess.engine.Cp(0)) if score.is_mate(): value = 1.0 if score.mate() > 0 else -1.0 else: cp = score.relative.score(mate_score=10000) value = max(-1, min(1, cp / 1000)) raw_values.append(value) action_labels.append(move.uci()) move_type = 'quiet' if board.is_capture(move): move_type = 'capture' elif board.gives_check(move): move_type = 'check' elif board.is_castling(move): move_type = 'castle' candidates.append(MoveCandidate( move=move.uci(), prob=0, value=value, from_sq=move.from_square, to_sq=move.to_square, is_capture=board.is_capture(move), is_check=board.gives_check(move), move_type=move_type )) # Softmax probabilities values_arr = np.array(raw_values) exp_vals = np.exp((values_arr - values_arr.max()) * 3) probs = exp_vals / exp_vals.sum() for i, c in enumerate(candidates): c.prob = float(probs[i]) # Build imagination (predicted opponent responses) imagination = {} for i, c in enumerate(candidates[:3]): test_board = board.copy() test_board.push(chess.Move.from_uci(c.move)) if not test_board.is_game_over(): resp_info = engine_analyse(test_board, chess.engine.Limit(depth=8), multipv=1) if resp_info: resp_move = resp_info[0]['pv'][0].uci() resp_score = resp_info[0].get('score', chess.engine.Cp(0)) if resp_score.is_mate(): resp_val = -1.0 if resp_score.mate() > 0 else 1.0 else: cp = resp_score.relative.score(mate_score=10000) resp_val = -max(-1, min(1, cp / 1000)) imagination[i] = { 'response': resp_move, 'value': resp_val, 'line': f"{c.move} → {resp_move}" } # Build reasoning reasoning = [] if candidates[0].is_capture: reasoning.append("Top move captures material") if candidates[0].is_check: reasoning.append("Top move gives check") if features['material'] > 2: reasoning.append("White has material advantage (+{:.0f})".format(features['material'])) elif features['material'] < -2: reasoning.append("Black has material advantage ({:.0f})".format(features['material'])) if features['center_control'] > 0.5: reasoning.append("Strong center control") if len(candidates) > 1 and abs(candidates[0].value - candidates[1].value) < 0.1: reasoning.append("Multiple strong options") if features['white_king_danger'] > 0: reasoning.append("⚠ White king under attack") if features['black_king_danger'] > 0: reasoning.append("Black king vulnerable") hold_data = { 'action_probs': [c.prob for c in candidates], 'action_labels': action_labels, 'value': float(candidates[0].value) if candidates else 0, 'features': features, 'reasoning': reasoning, 'imagination': imagination, 'ai_choice': 0, 'ai_confidence': float(probs[0]) if len(probs) > 0 else 0 } return candidates, hold_data # ═══════════════════════════════════════════════════════════════════════════════ # DASH APP # ═══════════════════════════════════════════════════════════════════════════════ app = dash.Dash(__name__, suppress_callback_exceptions=True) DARK_BG = '#0a0a0f' PANEL_BG = '#0d0d12' ACCENT = '#00FF88' ACCENT_DIM = '#00FF8844' app.layout = html.Div([ # Stores dcc.Store(id='board-fen', data=chess.STARTING_FEN), dcc.Store(id='candidates-store', data=[]), dcc.Store(id='hold-data-store', data={}), dcc.Store(id='selected-idx', data=0), dcc.Store(id='is-held', data=False), dcc.Store(id='move-history', data=[]), dcc.Store(id='game-status', data='playing'), # playing, white_wins, black_wins, draw # Header html.Div([ html.H1("CASCADE-LATTICE Chess", style={ 'color': ACCENT, 'fontFamily': 'monospace', 'margin': '0', 'fontSize': '24px', 'letterSpacing': '2px' }), html.Div("Human-AI Decision Support Demo", style={ 'color': '#666', 'fontFamily': 'monospace', 'fontSize': '12px' }) ], style={'textAlign': 'center', 'padding': '15px', 'borderBottom': f'1px solid {ACCENT_DIM}'}), # Main content html.Div([ # LEFT: 3D Board + Controls html.Div([ # Control buttons html.Div([ html.Button("⏸ HOLD", id='btn-hold', style={ 'margin': '5px', 'padding': '10px 20px', 'fontSize': '14px', 'fontFamily': 'monospace', 'fontWeight': 'bold', 'backgroundColor': ACCENT_DIM, 'color': ACCENT, 'border': f'2px solid {ACCENT}', 'borderRadius': '6px', 'cursor': 'pointer' }), html.Button("↻ RESET", id='btn-reset', style={ 'margin': '5px', 'padding': '10px 20px', 'fontSize': '14px', 'fontFamily': 'monospace', 'fontWeight': 'bold', 'backgroundColor': '#FF444422', 'color': '#FF4444', 'border': '2px solid #FF4444', 'borderRadius': '6px', 'cursor': 'pointer' }) ], style={'textAlign': 'center', 'padding': '10px'}), # 3D iframe html.Iframe( id='chess-3d-iframe', src='/assets/chess3d.html', style={'width': '100%', 'height': '500px', 'border': 'none', 'borderRadius': '8px'} ), html.Div(id='scene-state', style={'display': 'none'}), # Move selection (when HELD) html.Div([ html.Div(id='move-buttons', style={'textAlign': 'center', 'padding': '10px'}), html.Div([ html.Button("✓ COMMIT", id='btn-commit', style={ 'display': 'none', 'margin': '10px auto', 'padding': '15px 40px', 'fontSize': '16px', 'fontFamily': 'monospace', 'fontWeight': 'bold', 'backgroundColor': ACCENT_DIM, 'color': ACCENT, 'border': f'2px solid {ACCENT}', 'borderRadius': '8px', 'cursor': 'pointer' }) ], style={'textAlign': 'center'}) ]), # Status html.Div(id='game-status-display', style={ 'textAlign': 'center', 'padding': '10px', 'fontFamily': 'monospace', 'color': '#888', 'fontSize': '14px' }) ], style={'flex': '1', 'minWidth': '450px', 'padding': '10px'}), # RIGHT: Cascade Info Panel html.Div([ html.Div([ html.Span("◈ ", style={'color': ACCENT}), html.Span("INFORMATIONAL WEALTH", style={'color': '#888'}) ], style={'fontFamily': 'monospace', 'fontSize': '14px', 'marginBottom': '15px', 'borderBottom': f'1px solid {ACCENT_DIM}', 'paddingBottom': '10px'}), html.Div(id='wealth-panel', style={'fontFamily': 'monospace', 'fontSize': '12px'}) ], style={ 'flex': '1', 'minWidth': '350px', 'padding': '15px', 'backgroundColor': PANEL_BG, 'borderRadius': '8px', 'margin': '10px', 'border': f'1px solid {ACCENT_DIM}' }) ], style={'display': 'flex', 'flexWrap': 'wrap', 'justifyContent': 'center'}), # Move history html.Div([ html.Div("MOVE HISTORY", style={'color': '#666', 'fontSize': '10px', 'marginBottom': '5px'}), html.Div(id='history-display', style={'color': '#888', 'fontSize': '11px', 'maxHeight': '60px', 'overflow': 'auto'}) ], style={'fontFamily': 'monospace', 'textAlign': 'center', 'padding': '10px'}) ], style={'backgroundColor': DARK_BG, 'minHeight': '100vh', 'color': '#fff'}) # ═══════════════════════════════════════════════════════════════════════════════ # CALLBACKS # ═══════════════════════════════════════════════════════════════════════════════ # HOLD button - get candidates @callback( Output('candidates-store', 'data'), Output('hold-data-store', 'data'), Output('is-held', 'data'), Output('selected-idx', 'data'), Input('btn-hold', 'n_clicks'), State('board-fen', 'data'), State('is-held', 'data'), State('game-status', 'data'), prevent_initial_call=True ) def on_hold_click(n, fen, is_held, status): if status != 'playing': raise dash.exceptions.PreventUpdate if is_held: # Un-hold return [], {}, False, 0 else: # Hold - get candidates board = chess.Board(fen) if board.turn != chess.WHITE: # Not white's turn raise dash.exceptions.PreventUpdate candidates, hold_data = get_candidates_with_hold(board) if not candidates: raise dash.exceptions.PreventUpdate return [c.__dict__ for c in candidates], hold_data, True, 0 # RESET button @callback( Output('board-fen', 'data', allow_duplicate=True), Output('candidates-store', 'data', allow_duplicate=True), Output('hold-data-store', 'data', allow_duplicate=True), Output('is-held', 'data', allow_duplicate=True), Output('move-history', 'data', allow_duplicate=True), Output('game-status', 'data', allow_duplicate=True), Input('btn-reset', 'n_clicks'), prevent_initial_call=True ) def on_reset(n): return chess.STARTING_FEN, [], {}, False, [], 'playing' # Move button clicks - select candidate @callback( Output('selected-idx', 'data', allow_duplicate=True), Input({'type': 'move-btn', 'index': dash.ALL}, 'n_clicks'), prevent_initial_call=True ) def on_move_select(clicks): if not clicks or not any(clicks): raise dash.exceptions.PreventUpdate triggered = ctx.triggered_id if triggered and 'index' in triggered: return triggered['index'] raise dash.exceptions.PreventUpdate # COMMIT - execute move + opponent responds @callback( Output('board-fen', 'data', allow_duplicate=True), Output('candidates-store', 'data', allow_duplicate=True), Output('hold-data-store', 'data', allow_duplicate=True), Output('is-held', 'data', allow_duplicate=True), Output('move-history', 'data', allow_duplicate=True), Output('game-status', 'data', allow_duplicate=True), Input('btn-commit', 'n_clicks'), State('board-fen', 'data'), State('candidates-store', 'data'), State('selected-idx', 'data'), State('move-history', 'data'), prevent_initial_call=True ) def on_commit(n, fen, candidates_data, selected_idx, history): if not n or not candidates_data: raise dash.exceptions.PreventUpdate board = chess.Board(fen) status = 'playing' # 1. Execute White's move if selected_idx < len(candidates_data): move_uci = candidates_data[selected_idx]['move'] move = chess.Move.from_uci(move_uci) board.push(move) history = history + [move_uci] print(f"[GAME] White plays: {move_uci}") # Check game over after white if board.is_game_over(): if board.is_checkmate(): status = 'white_wins' else: status = 'draw' return board.fen(), [], {}, False, history, status # 2. Engine plays Black result = engine_play(board, chess.engine.Limit(depth=12)) if result and result.move: board.push(result.move) history = history + [result.move.uci()] print(f"[GAME] Black plays: {result.move.uci()}") # Check game over after black if board.is_game_over(): if board.is_checkmate(): status = 'black_wins' else: status = 'draw' return board.fen(), [], {}, False, history, status # Show/hide COMMIT button @callback( Output('btn-commit', 'style'), Input('is-held', 'data') ) def toggle_commit(is_held): base = { 'margin': '10px auto', 'padding': '15px 40px', 'fontSize': '16px', 'fontFamily': 'monospace', 'fontWeight': 'bold', 'backgroundColor': ACCENT_DIM, 'color': ACCENT, 'border': f'2px solid {ACCENT}', 'borderRadius': '8px', 'cursor': 'pointer' } base['display'] = 'block' if is_held else 'none' return base # Move buttons @callback( Output('move-buttons', 'children'), Input('candidates-store', 'data'), Input('selected-idx', 'data'), Input('is-held', 'data') ) def render_move_buttons(candidates_data, selected_idx, is_held): if not is_held or not candidates_data: return [] buttons = [] for i, c in enumerate(candidates_data): is_selected = (i == selected_idx) style = { 'margin': '5px', 'padding': '10px 15px', 'fontSize': '13px', 'fontFamily': 'monospace', 'fontWeight': 'bold', 'borderRadius': '6px', 'cursor': 'pointer', 'transition': 'all 0.2s' } if is_selected: style['backgroundColor'] = ACCENT style['color'] = '#000' style['border'] = f'2px solid {ACCENT}' style['boxShadow'] = f'0 0 15px {ACCENT}' else: style['backgroundColor'] = '#1a1a24' style['color'] = '#888' style['border'] = '2px solid #333' prob_pct = c['prob'] * 100 label = f"{c['move']} ({prob_pct:.0f}%)" buttons.append(html.Button( label, id={'type': 'move-btn', 'index': i}, style=style )) return buttons # Wealth panel @callback( Output('wealth-panel', 'children'), Input('hold-data-store', 'data'), Input('selected-idx', 'data'), Input('is-held', 'data') ) def render_wealth(hold_data, selected_idx, is_held): if not is_held or not hold_data: return html.Div("Press HOLD to analyze position", style={'color': '#555', 'fontStyle': 'italic'}) features = hold_data.get('features', {}) reasoning = hold_data.get('reasoning', []) imagination = hold_data.get('imagination', {}) probs = hold_data.get('action_probs', []) labels = hold_data.get('action_labels', []) sections = [] # Position Features sections.append(html.Div([ html.Div("▸ POSITION FEATURES", style={'color': ACCENT, 'marginBottom': '8px'}), html.Div([ html.Div(f"Material: {features.get('material', 0):+.1f}", style={'color': '#aaa'}), html.Div(f"Center: {features.get('center_control', 0):+.2f}", style={'color': '#aaa'}), html.Div(f"Phase: {features.get('phase', '?')}", style={'color': '#aaa'}), html.Div(f"W mobility: {features.get('white_mobility', 0)}", style={'color': '#aaa'}), html.Div(f"B mobility: {features.get('black_mobility', 0)}", style={'color': '#aaa'}), ], style={'paddingLeft': '15px', 'marginBottom': '15px'}) ])) # AI Reasoning if reasoning: sections.append(html.Div([ html.Div("▸ AI REASONING", style={'color': ACCENT, 'marginBottom': '8px'}), html.Div([ html.Div(f"• {r}", style={'color': '#aaa', 'marginBottom': '3px'}) for r in reasoning ], style={'paddingLeft': '15px', 'marginBottom': '15px'}) ])) # Imagination (predicted lines) if imagination: sections.append(html.Div([ html.Div("▸ IMAGINATION (predicted responses)", style={'color': ACCENT, 'marginBottom': '8px'}), html.Div([ html.Div([ html.Span(f"#{int(k)+1}: ", style={'color': '#666'}), html.Span(img['line'], style={'color': '#aaa'}), html.Span(f" ({img['value']:+.2f})", style={'color': '#666'}) ], style={'marginBottom': '3px'}) for k, img in imagination.items() ], style={'paddingLeft': '15px', 'marginBottom': '15px'}) ])) # Selected move confidence if selected_idx < len(probs): conf = probs[selected_idx] * 100 move = labels[selected_idx] if selected_idx < len(labels) else '?' sections.append(html.Div([ html.Div("▸ SELECTED MOVE", style={'color': ACCENT, 'marginBottom': '8px'}), html.Div([ html.Span(move, style={'color': '#fff', 'fontSize': '18px', 'fontWeight': 'bold'}), html.Span(f" {conf:.1f}% confidence", style={'color': '#888'}) ], style={'paddingLeft': '15px'}) ])) return sections # Scene state for Three.js @callback( Output('scene-state', 'children'), Input('board-fen', 'data'), Input('candidates-store', 'data'), Input('selected-idx', 'data') ) def update_scene(fen, candidates_data, selected_idx): candidates = [] if candidates_data: for i, c in enumerate(candidates_data): candidates.append({ 'from_sq': c['from_sq'], 'to_sq': c['to_sq'], 'prob': c['prob'], 'selected': (i == selected_idx) }) state = {'fen': fen, 'candidates': candidates} return json.dumps(state) # Game status display @callback( Output('game-status-display', 'children'), Input('game-status', 'data'), Input('board-fen', 'data') ) def show_status(status, fen): board = chess.Board(fen) turn = "White" if board.turn == chess.WHITE else "Black" if status == 'white_wins': return html.Span("✓ WHITE WINS!", style={'color': ACCENT, 'fontWeight': 'bold'}) elif status == 'black_wins': return html.Span("✗ BLACK WINS", style={'color': '#FF4444', 'fontWeight': 'bold'}) elif status == 'draw': return html.Span("½ DRAW", style={'color': '#888'}) else: return f"{turn} to move • Press HOLD to analyze" # History display @callback( Output('history-display', 'children'), Input('move-history', 'data') ) def show_history(history): if not history: return "Game start" moves = [] for i in range(0, len(history), 2): num = i // 2 + 1 white = history[i] black = history[i + 1] if i + 1 < len(history) else "..." moves.append(f"{num}. {white} {black}") return " ".join(moves) # ═══════════════════════════════════════════════════════════════════════════════ # MAIN # ═══════════════════════════════════════════════════════════════════════════════ if __name__ == '__main__': print("\n" + "=" * 50) print("CASCADE-LATTICE Chess (Clean Build)") print("=" * 50) print("Open: http://127.0.0.1:8050") print("=" * 50 + "\n") app.run(debug=False, host='127.0.0.1', port=8050)