huzzle-labs/visual_memory

1"""Standalone play server for manual Phantom Grid gameplay.2 3Completely separate from the OpenEnv app.py — does NOT affect4HuggingFace deployment, Docker builds, or run_eval.py in any way.5 6Runs on port 8001 by default. Uses the game engine and renderer directly.7 8Usage:9    cd visual-memory10    python play_server.py11    # Then open play.html in a browser12"""13 14from __future__ import annotations15 16import json17import os18import sys19from pathlib import Path20 21import uvicorn22from fastapi import FastAPI23from fastapi.middleware.cors import CORSMiddleware24from fastapi.responses import HTMLResponse, FileResponse25from pydantic import BaseModel26 27sys.path.insert(0, str(Path(__file__).resolve().parent))28 29from server.engine import GameEngine30from server.renderer import Renderer31 32SCENARIOS_DIR = os.path.join(os.path.dirname(__file__), "scenarios")33 34app = FastAPI(title="Phantom Grid — Play Server")35app.add_middleware(36    CORSMiddleware,37    allow_origins=["*"],38    allow_methods=["*"],39    allow_headers=["*"],40)41 42engine: GameEngine | None = None43renderer = Renderer()44 45 46def _load_scenario_file(scenario_id: str) -> dict:47    path = os.path.join(SCENARIOS_DIR, f"{scenario_id}.json")48    if not os.path.isfile(path):49        raise FileNotFoundError(f"Scenario '{scenario_id}' not found at {path}")50    with open(path) as f:51        return json.load(f)52 53 54def _board_response() -> dict:55    """Build a unified response with board SVG + game status."""56    if engine is None:57        return {"error": "No scenario loaded."}58    bs = engine.get_board_state()59    view = renderer.get_board_view(60        bs.visible_cells, bs.board_width, bs.board_height,61        scenario_type=bs.scenario_type, step_count=bs.step_count,62    )63    status = engine.get_status()64    return {"board": view, "status": status}65 66 67@app.get("/")68async def index():69    html_path = os.path.join(os.path.dirname(__file__), "play.html")70    if os.path.isfile(html_path):71        return FileResponse(html_path, media_type="text/html")72    return HTMLResponse("<h1>play.html not found</h1>", status_code=404)73 74 75@app.get("/scenarios")76async def list_scenarios():77    results = []78    for fname in sorted(os.listdir(SCENARIOS_DIR)):79        if not fname.endswith(".json"):80            continue81        try:82            data = _load_scenario_file(fname.replace(".json", ""))83            results.append({84                "scenario_id": data.get("scenario_id", fname.replace(".json", "")),85                "type": data.get("type", "hidden_grid"),86                "difficulty": data.get("difficulty", "hard"),87                "board_size": f"{data.get('board_width', '?')}x{data.get('board_height', '?')}",88                "description": data.get("description", ""),89            })90        except Exception:91            continue92    return {"scenarios": results}93 94 95class LoadReq(BaseModel):96    scenario_id: str97 98@app.post("/load")99async def load_scenario(req: LoadReq):100    global engine101    try:102        data = _load_scenario_file(req.scenario_id)103    except FileNotFoundError as e:104        return {"error": str(e)}105    engine = GameEngine(data)106    resp = _board_response()107    resp["loaded"] = True108    resp["how_to_play"] = data.get("how_to_play", "")109    resp["scenario_description"] = data.get("description", "")110    return resp111 112 113class CellReq(BaseModel):114    row: int115    col: int116 117@app.post("/reveal")118async def reveal(req: CellReq):119    if engine is None:120        return {"error": "No scenario loaded."}121    result = engine.reveal_cell(req.row, req.col)122    resp = _board_response()123    resp["action_result"] = result124    return resp125 126 127@app.post("/flag")128async def flag(req: CellReq):129    if engine is None:130        return {"error": "No scenario loaded."}131    result = engine.flag_cell(req.row, req.col)132    resp = _board_response()133    resp["action_result"] = result134    return resp135 136 137@app.post("/unflag")138async def unflag(req: CellReq):139    if engine is None:140        return {"error": "No scenario loaded."}141    result = engine.unflag_cell(req.row, req.col)142    resp = _board_response()143    resp["action_result"] = result144    return resp145 146 147@app.post("/move_viewport")148async def move_viewport(req: CellReq):149    if engine is None:150        return {"error": "No scenario loaded."}151    result = engine.move_viewport(req.row, req.col)152    resp = _board_response()153    resp["action_result"] = result154    return resp155 156 157class InspectReq(BaseModel):158    center_row: int159    center_col: int160    radius: int = 1161 162@app.post("/inspect")163async def inspect(req: InspectReq):164    if engine is None:165        return {"error": "No scenario loaded."}166    if engine.game_over:167        return {"error": "Game is already over."}168    if req.radius < 1 or req.radius > 3:169        return {"error": "Radius must be between 1 and 3."}170 171    engine.step_count += 1172    engine._tick_pattern_memory()173 174    visible = engine.get_visible_board()175    region = []176    for r in range(max(0, req.center_row - req.radius),177                    min(engine.height, req.center_row + req.radius + 1)):178        for c in range(max(0, req.center_col - req.radius),179                        min(engine.width, req.center_col + req.radius + 1)):180            cell = visible[r][c]181            region.append({"row": r, "col": c, "state": cell["state"], "content": cell.get("content")})182 183    resp = _board_response()184    resp["action_result"] = {"cells": region}185    return resp186 187 188@app.get("/status")189async def status():190    if engine is None:191        return {"error": "No scenario loaded."}192    return engine.get_status()193 194 195@app.get("/board")196async def board():197    return _board_response()198 199 200@app.get("/recall")201async def recall():202    if engine is None:203        return {"error": "No scenario loaded."}204    bs = engine.get_board_state()205    return {206        "discovered_signals": bs.discovered_signals,207        "memory_events": bs.memory_events,208    }209 210 211class SubmitReq(BaseModel):212    flagged_positions: list[list[int]] = []213    safe_positions: list[list[int]] = []214 215@app.post("/submit")216async def submit(req: SubmitReq):217    if engine is None:218        return {"error": "No scenario loaded."}219    result = engine.submit_solution(220        flagged_positions=req.flagged_positions,221        safe_positions=req.safe_positions,222    )223    resp = _board_response()224    resp["action_result"] = result225    return resp226 227 228if __name__ == "__main__":229    uvicorn.run(app, host="0.0.0.0", port=8001)230