Update app.py

app.py

@@ -882,202 +882,233 @@ def scrape_and_retrieve(url: str, query: str) -> str:
         except Exception as e:
             return f"Error processing page: {str(e)}\n{traceback.format_exc()}"
             
-def analyze_chess_position(args: str, state: AgentState) -> str:
+class ChessAnalysisInput(BaseModel):
+    image_path: str = Field(description="Path to chess board image file")
+    description: str = Field(description="Any additional context about the position (optional)", default="")
+
+@tool(args_schema=ChessAnalysisInput)
+def analyze_chess_position(image_path: str, description: str = "") -> str:
     """
-    Analyze chess position using Stockfish engine via lichess API or python-chess
-    Input format: "image_path|description" or just FEN notation
+    Analyzes a chess position from an image and returns the best move.
+    
+    Use this tool when:
+    - Question mentions chess, checkmate, or chess notation
+    - An image file shows a chess board
+    - Need to find the best move in a position
+    
+    The tool will:
+    1. Extract the position from the image using Gemini Vision
+    2. Analyze it using Lichess cloud API
+    3. Return the best move in algebraic notation (e.g., "Qh5" or "Nf6")
+    
+    Example: analyze_chess_position(image_path="/tmp/chess_board.png")
     """
+    if not image_path:
+        return "Error: image_path is required."
+    
+    print(f"♟️ Analyzing chess position from: {image_path}")
+    
+    # Find the file
+    chess_image = find_file(image_path)
+    
+    # If not found via find_file, try direct path (for /tmp files on HF)
+    if not chess_image and os.path.exists(image_path):
+        chess_image = Path(image_path)
+    
+    if not chess_image or not chess_image.exists():
+        return f"Error: Chess board image not found at '{image_path}'. Check the [FILE ATTACHED: ...] path in the question."
+    
+    print(f"✓ Found chess image at: {chess_image}")
+    
     try:
-        # Try to use python-chess with Stockfish
+        # Step 1: Extract FEN notation from image using Gemini
+        GOOGLE_API_KEY = os.getenv("GEMINI_API_KEY")
+        if not GOOGLE_API_KEY:
+            return "Error: GEMINI_API_KEY not set in Space secrets."
+        
+        print("📸 Extracting chess position from image using Gemini...")
+        
         try:
-            import chess
-            import chess.engine
+            # Load image
+            img = Image.open(chess_image)
+            print(f"   Image loaded: {img.size}, mode: {img.mode}")
+            
+            # Convert to RGB if needed
+            if img.mode not in ['RGB', 'RGBA']:
+                img = img.convert('RGB')
+            
+            # Encode to base64
+            buffered = io.BytesIO()
+            img.save(buffered, format="JPEG")
+            img_base64 = base64.b64encode(buffered.getvalue()).decode()
+            
+            # Use Gemini Vision to extract FEN
+            vision_llm = ChatGoogleGenerativeAI(
+                model="gemini-2.0-flash-exp",
+                google_api_key=GOOGLE_API_KEY,
+                temperature=0
+            )
             
-            # Check if we have an image to analyze first
-            if '|' in args and os.path.exists(args.split('|')[0]):
-                image_path = args.split('|')[0]
-                
-                # Use Gemini to extract FEN from image
-                print("📸 Extracting chess position from image...")
-                img = Image.open(image_path)
-                model = genai.GenerativeModel('gemini-2.0-flash-exp')
-                
-                fen_prompt = """Analyze this chess board image and provide the position in FEN notation.
-                
-                Important instructions:
-                1. Carefully identify each piece and its position
-                2. Determine whose turn it is (look for indicators in the image)
-                3. Return ONLY the FEN string, nothing else
-                4. Format: piece_placement active_color castling en_passant halfmove fullmove
-                
-                Example: rnbqkbnr/pppppppp/8/8/8/8/PPPPPPPP/RNBQKBNR w KQkq - 0 1
-                
-                If it says "Black to move" or "Black's turn", use 'b' for active color.
-                If it says "White to move" or "White's turn", use 'w' for active color."""
-                
-                response = model.generate_content([fen_prompt, img])
-                fen = response.text.strip()
+            fen_prompt = """Analyze this chess board image and provide the position in FEN notation.
+
+CRITICAL INSTRUCTIONS:
+1. Carefully identify each piece (uppercase for White: K, Q, R, B, N, P; lowercase for Black: k, q, r, b, n, p)
+2. Read the board from rank 8 (top) to rank 1 (bottom), left to right
+3. Determine whose turn it is:
+   - If text says "Black to move" or "Black's turn" → use 'b'
+   - If text says "White to move" or "White's turn" → use 'w'
+   - If no indication, analyze the position and make your best guess
+4. Return ONLY the FEN string in this exact format:
+   piece_placement active_color castling en_passant halfmove fullmove
+
+Example FEN: rnbqkbnr/pppppppp/8/8/8/8/PPPPPPPP/RNBQKBNR w KQkq - 0 1
+
+For piece placement:
+- Use numbers for empty squares (1-8)
+- Use '/' to separate ranks
+- White pieces: UPPERCASE (K, Q, R, B, N, P)
+- Black pieces: lowercase (k, q, r, b, n, p)
+
+Return ONLY the FEN string, nothing else."""
+
+            message = HumanMessage(
+                content=[
+                    {"type": "text", "text": fen_prompt},
+                    {
+                        "type": "image_url",
+                        "image_url": f"data:image/jpeg;base64,{img_base64}"
+                    }
+                ]
+            )
+            
+            response = vision_llm.invoke([message])
+            fen_raw = response.content.strip()
+            
+            print(f"📝 Raw FEN response: {fen_raw}")
+            
+            # Clean up FEN (remove markdown, explanations, etc.)
+            fen = None
+            fen_lines = fen_raw.split('\n')
+            
+            for line in fen_lines:
+                line = line.strip()
+                # Remove markdown code fences
+                line = line.replace('```', '').replace('fen', '').strip()
                 
-                # Clean up the FEN (remove markdown, explanations, etc.)
-                fen_lines = fen.split('\n')
-                for line in fen_lines:
-                    line = line.strip()
-                    # FEN should have spaces and slashes
-                    if '/' in line and ' ' in line and not line.startswith('#'):
+                # FEN should have '/' for ranks and spaces for components
+                if '/' in line and ' ' in line and not line.startswith('#'):
+                    # Basic validation: should have pieces or numbers, slashes, and spaces
+                    if any(c in line for c in 'kqrbnpKQRBNP12345678'):
                         fen = line
                         break
-                
-                print(f"📊 Extracted FEN: {fen}")
-            else:
-                # Direct FEN input
-                fen = args.strip()
             
-            # Parse the position
+            if not fen:
+                # Try to extract from any line with slash and space
+                for line in fen_lines:
+                    if '/' in line and ' ' in line:
+                        fen = line.strip()
+                        break
+            
+            if not fen:
+                return f"Error: Could not extract valid FEN notation from image. Gemini response: {fen_raw[:200]}"
+            
+            print(f"✓ Extracted FEN: {fen}")
+            
+        except Exception as e:
+            return f"Error extracting position from image: {str(e)}\n{traceback.format_exc()}"
+        
+        # Step 2: Analyze position using Lichess cloud API
+        print("🔍 Analyzing position with Lichess cloud engine...")
+        
+        try:
+            # Try python-chess for better move notation (optional)
             try:
-                board = chess.Board(fen)
-            except Exception as e:
-                return f"N/A_REQUIRED: Invalid FEN notation - {str(e)}"
-            
-            # Try to use Stockfish engine
-            stockfish_paths = [
-                "/usr/games/stockfish",
-                "/usr/local/bin/stockfish",
-                "/opt/homebrew/bin/stockfish",
-                "stockfish",
-                "./stockfish"
-            ]
+                import chess
+                use_python_chess = True
+                print("✓ Using python-chess for move conversion")
+            except ImportError:
+                use_python_chess = False
+                print("⚠️ python-chess not available, using UCI notation")
+            
+            # Query Lichess cloud evaluation
+            lichess_url = "https://lichess.org/api/cloud-eval"
+            params = {
+                "fen": fen.strip(),
+                "multiPv": 1  # Get only the best move
+            }
             
-            engine_path = None
-            for path in stockfish_paths:
-                if os.path.exists(path) or path == "stockfish":
-                    engine_path = path
-                    break
-            
-            if not engine_path:
-                # Fallback to lichess API
-                print("⚠️ Stockfish not found locally, using Lichess API...")
-                return analyze_chess_via_lichess(board.fen(), state)
-            
-            # Use local Stockfish
-            with chess.engine.SimpleEngine.popen_uci(engine_path) as engine:
-                # Analyze position
-                info = engine.analyse(board, chess.engine.Limit(depth=20))
-                best_move = info.get("pv")[0] if "pv" in info else None
-                
-                if best_move:
-                    # Convert to algebraic notation
-                    san_move = board.san(best_move)
-                    
-                    # Get evaluation score
-                    score = info.get("score")
-                    score_str = ""
-                    if score:
-                        if score.is_mate():
-                            mate_in = score.relative.moves
-                            score_str = f" (Mate in {abs(mate_in)})"
-                        else:
-                            cp = score.relative.score()
-                            score_str = f" (Eval: {cp/100:.2f})"
+            response = requests.get(lichess_url, params=params, timeout=15)
+            
+            if response.status_code != 200:
+                return f"Error: Lichess API returned status {response.status_code}. Position may not be in cloud database."
+            
+            data = response.json()
+            print(f"📊 Lichess response: {data}")
+            
+            # Extract best move
+            if "pvs" not in data or len(data["pvs"]) == 0:
+                return "Error: Position not found in Lichess cloud database. Try a different position or check if the FEN is valid."
+            
+            best_pv = data["pvs"][0]
+            moves = best_pv.get("moves", "").split()
+            
+            if not moves:
+                return "Error: No moves found in Lichess analysis."
+            
+            best_move_uci = moves[0]
+            print(f"🎯 Best move (UCI): {best_move_uci}")
+            
+            # Convert UCI to Standard Algebraic Notation (SAN)
+            if use_python_chess:
+                try:
+                    board = chess.Board(fen)
+                    uci_move = chess.Move.from_uci(best_move_uci)
+                    san_move = board.san(uci_move)
                     
-                    # Check if this move leads to checkmate
-                    board_copy = board.copy()
-                    board_copy.push(best_move)
+                    # Get evaluation
+                    cp = best_pv.get("cp")  # centipawns
+                    mate = best_pv.get("mate")  # mate in X moves
                     
-                    result = f"{san_move}{score_str}"
+                    if mate is not None:
+                        eval_str = f" (Mate in {abs(mate)})"
+                    elif cp is not None:
+                        eval_str = f" (Eval: {cp/100:+.2f})"
+                    else:
+                        eval_str = ""
                     
-                    if board_copy.is_checkmate():
-                        result += " - Checkmate!"
-                    elif board_copy.is_check():
-                        result += " - Check"
+                    # Check if move leads to check/checkmate
+                    board.push(uci_move)
+                    if board.is_checkmate():
+                        check_str = " - Checkmate!"
+                    elif board.is_check():
+                        check_str = " - Check"
+                    else:
+                        check_str = ""
                     
-                    print(f"♟️ Best move: {result}")
-                    return result
-                else:
-                    return "N/A_REQUIRED: Could not determine best move"
+                    final_result = f"{san_move}{eval_str}{check_str}"
+                    print(f"✅ Best move: {final_result}")
+                    return final_result
                     
-        except ImportError:
-            print("⚠️ python-chess not installed, using Lichess API...")
-            # Extract FEN from image if needed
-            if '|' in args and os.path.exists(args.split('|')[0]):
-                image_path = args.split('|')[0]
-                img = Image.open(image_path)
-                model = genai.GenerativeModel('gemini-2.0-flash-exp')
-                
-                response = model.generate_content([
-                    "Extract the chess position in FEN notation. Return ONLY the FEN string.",
-                    img
-                ])
-                fen = response.text.strip()
+                except Exception as e:
+                    print(f"⚠️ Could not convert to SAN: {e}")
+                    # Fall back to UCI notation
+                    return best_move_uci
             else:
-                fen = args.strip()
-            
-            return analyze_chess_via_lichess(fen, state)
-            
-    except Exception as e:
-        state.add_failure('chess', str(e))
-        return f"N/A_REQUIRED: Chess analysis failed - {str(e)}"
-
-
-def analyze_chess_via_lichess(fen: str, state: AgentState) -> str:
-    """
-    Analyze chess position using Lichess cloud API
-    """
-    try:
-        # Lichess cloud evaluation API
-        url = "https://lichess.org/api/cloud-eval"
-        
-        # Clean FEN
-        fen = fen.strip().replace('```', '').replace('fen', '').strip()
-        
-        params = {
-            "fen": fen,
-            "multiPv": 1  # Get best move only
-        }
-        
-        response = requests.get(url, params=params, timeout=10)
-        
-        if response.status_code == 200:
-            data = response.json()
-            
-            if "pvs" in data and len(data["pvs"]) > 0:
-                best_pv = data["pvs"][0]
+                # Return UCI notation if python-chess not available
+                return best_move_uci
                 
-                # Get the moves in UCI notation
-                moves = best_pv.get("moves", "").split()
-                if moves:
-                    # Convert UCI to SAN using python-chess if available
-                    try:
-                        import chess
-                        board = chess.Board(fen)
-                        uci_move = chess.Move.from_uci(moves[0])
-                        san_move = board.san(uci_move)
-                        
-                        # Get evaluation
-                        cp = best_pv.get("cp")
-                        mate = best_pv.get("mate")
-                        
-                        if mate is not None:
-                            eval_str = f" (Mate in {abs(mate)})"
-                        elif cp is not None:
-                            eval_str = f" (Eval: {cp/100:.2f})"
-                        else:
-                            eval_str = ""
-                        
-                        return f"{san_move}{eval_str}"
-                    except:
-                        # Return UCI move if can't convert
-                        return moves[0]
-                else:
-                    return "N/A_REQUIRED: No moves found in analysis"
-            else:
-                return "N/A_REQUIRED: Position not in Lichess cloud database"
-        else:
-            state.add_failure('lichess', f'HTTP {response.status_code}')
-            return f"N/A_REQUIRED: Lichess API error {response.status_code}"
+        except requests.Timeout:
+            return "Error: Lichess API request timed out. Try again."
+        except requests.RequestException as e:
+            return f"Error querying Lichess API: {str(e)}"
+        except Exception as e:
+            return f"Error analyzing position: {str(e)}\n{traceback.format_exc()}"
             
     except Exception as e:
-        state.add_failure('lichess', str(e))
-        return f"N/A_REQUIRED: Lichess analysis failed - {str(e)}"
+        error_msg = f"Chess analysis failed: {str(e)}"
+        print(f"❌ {error_msg}")
+        print(traceback.format_exc())
+        return error_msg
 
 class FinalAnswerInput(BaseModel):
     answer: str = Field(description="Final answer - EXACTLY what was asked, nothing more")
@@ -1124,6 +1155,7 @@ defined_tools = [
     analyze_image,  
     get_youtube_transcript,
     scrape_and_retrieve,
+    analyze_chess_position,
     
     # Final
     final_answer_tool