Spaces:
Sleeping
Sleeping
File size: 5,788 Bytes
7a1529d 97c3bf5 d4b066d 7a1529d 97c3bf5 8d083d0 7a1529d 97c3bf5 7a1529d d4b066d 8d083d0 a1e07be 8c36a5e a1e07be 7a1529d 8d083d0 7a1529d 8d083d0 7a1529d 8d083d0 2968926 7a1529d 8d083d0 7a1529d 8d083d0 7a1529d 8d083d0 97c3bf5 b0a8dea 97c3bf5 b0a8dea 97c3bf5 b0a8dea 97c3bf5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 |
import io
import os
import tempfile
from fastapi import FastAPI, File, UploadFile, Form
from fastapi.responses import JSONResponse, StreamingResponse
from PIL import Image, UnidentifiedImageError
from routes.segmentation import segment_chess_board
from routes.detection import detect_pieces
from routes.fen_generator import gen_fen
from routes.chess_review import analyze_pgn
from typing import List, Dict, Any, Union
from pydantic import BaseModel
import asyncio
import sys
import tracemalloc
tracemalloc.start()
if sys.platform == "win32":
asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
app = FastAPI()
class DetectionResults(BaseModel):
boxes: list
confidences: list
classes: list
@app.get("/")
async def read_root():
return {
"name": "Narendra",
"age": 20,
"Gender": "Male"
}
@app.post("/getSeg")
async def get_seg(file: UploadFile = File(...)):
print(f'Image received: {file.filename}')
try:
image_content = await file.read()
if not image_content:
return JSONResponse(content={"error": "Empty file uploaded"}, status_code=400)
try:
image = Image.open(io.BytesIO(image_content))
except UnidentifiedImageError:
return JSONResponse(content={"error": "Invalid image format"}, status_code=400)
# If segment_chess_board is async, use `await`, otherwise remove `await`
segmented_image = await segment_chess_board(image)
if isinstance(segmented_image, dict):
return JSONResponse(content=segmented_image, status_code=400)
# Save to in-memory bytes
img_bytes = io.BytesIO()
segmented_image.save(img_bytes, format="PNG")
img_bytes.seek(0)
print("Image successfully processed and returned")
return StreamingResponse(
img_bytes,
media_type="image/png",
headers={"Content-Disposition": "inline; filename=output.png"}
)
except Exception as e:
return JSONResponse(content={"error": str(e)}, status_code=500)
@app.post("/getCoords")
async def get_coords(file: UploadFile = File(...)):
try:
image_content = await file.read()
if not image_content:
print("No image found")
return JSONResponse(content={"error": "Empty file uploaded"}, status_code=400)
try:
image = Image.open(io.BytesIO(image_content))
except UnidentifiedImageError:
return JSONResponse(content={"error": "Invalid image format"}, status_code=400)
detection_results = await detect_pieces(image)
if "error" in detection_results:
return JSONResponse(content=detection_results, status_code=400)
print("Image successfully processed and returned")
return JSONResponse(content={"detections": detection_results}, status_code=200)
except Exception as e:
print(f"Unexpected error: {str(e)}")
return JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)
@app.post("/getFen")
async def get_fen(file : UploadFile = File(), perspective : str = Form("w"), next_to_move : str = Form("w")):
if perspective not in ["w" , "b"]:
return JSONResponse(content={"error" : "Perspective should be w (white) or b (black)"}, status_code=500)
if next_to_move not in ["w" , "b"]:
return JSONResponse(content={"error" : "next to move should be w (white) or b (black)"}, status_code=500)
try:
image_content = await file.read()
if not image_content:
return JSONResponse(content={"error": "Empty file uploaded"}, status_code=400)
try:
image = Image.open(io.BytesIO(image_content))
except UnidentifiedImageError:
return JSONResponse(content={"error": "Invalid image format"}, status_code=400)
segmented_image = await segment_chess_board(image)
if isinstance(segmented_image, dict):
return JSONResponse(content=segmented_image, status_code=400)
segmented_image = segmented_image.resize((224, 224))
detection_results = await detect_pieces(segmented_image)
if "error" in detection_results:
return JSONResponse(content=detection_results, status_code=400)
fen = gen_fen(detection_results, perspective, next_to_move)
if not fen:
return JSONResponse(content={"error": "FEN generation failed", "details": "Invalid input data"}, status_code=500)
return JSONResponse(content={"FEN": fen}, status_code=200)
except Exception as e:
return JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)
@app.post('/getReview')
async def getReview(file: UploadFile = File(...)):
print(os.getcwd())
print("call recieved")
if not file.filename.endswith(".pgn"):
return JSONResponse(content={"error": "Invalid file format. Please upload a PGN file"}, status_code=400)
try:
# Save the uploaded file to a temporary file
with tempfile.NamedTemporaryFile(delete=False, suffix=".pgn") as tmp_file:
tmp_file.write(await file.read())
tmp_file_path = tmp_file.name
# Analyze the PGN file
analysis_result = analyze_pgn(tmp_file_path)
# Clean up the temporary file
os.remove(tmp_file_path)
if not analysis_result:
return JSONResponse(content={"error": "No game found in the PGN file"}, status_code=400)
return analysis_result
except Exception as e:
return JSONResponse(content={"error": "Unexpected error occurred", "details": str(e)}, status_code=500)
|