Spaces:
Sleeping
Sleeping
| import sys | |
| import os | |
| import uuid | |
| from typing import Dict, Any | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.staticfiles import StaticFiles | |
| from fastapi.responses import FileResponse | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel | |
| # Add parent directory to sys.path | |
| sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| from env import InteriorEnv | |
| from models import Action | |
| from tasks import TASKS | |
| app = FastAPI(title="Interior Design API") | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Static dir is one level up | |
| _static_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "static") | |
| if not os.path.exists(_static_dir): | |
| os.makedirs(_static_dir) | |
| app.mount("/static", StaticFiles(directory=_static_dir), name="static") | |
| async def serve_dashboard(): | |
| index_path = os.path.join(_static_dir, "index.html") | |
| if os.path.exists(index_path): | |
| return FileResponse(index_path) | |
| return {"message": "Welcome to Interior Env API"} | |
| class SessionStartReq(BaseModel): | |
| task_id: str | |
| class SessionStepReq(BaseModel): | |
| session_id: str | |
| action_type: str | |
| item: str | |
| x: int = 0 | |
| y: int = 0 | |
| rotation: int = 0 | |
| _sessions = {} | |
| async def get_tasks(): | |
| return {k: { | |
| "description": v["description"], | |
| "difficulty": v["difficulty"], | |
| "room_size": v["room_size"] | |
| } for k, v in TASKS.items()} | |
| # ── OpenEnv validator required endpoints ────────────────────────────────────── | |
| async def reset(): | |
| """Required by OpenEnv pre-validation script (validate-submission.sh).""" | |
| return {"status": "ok", "tasks": list(TASKS.keys())} | |
| async def health(): | |
| """Docker HEALTHCHECK support.""" | |
| return {"status": "healthy", "tasks": len(TASKS)} | |
| async def start_session(req: SessionStartReq): | |
| if req.task_id not in TASKS: | |
| raise HTTPException(status_code=400, detail="Invalid task ID") | |
| env = InteriorEnv() | |
| obs = env.reset(TASKS[req.task_id], req.task_id) | |
| sid = str(uuid.uuid4()) | |
| _sessions[sid] = {"env": env} | |
| return {"session_id": sid, "observation": obs.model_dump()} | |
| async def step_session(req: SessionStepReq): | |
| if req.session_id not in _sessions: | |
| raise HTTPException(status_code=400, detail="Invalid session ID") | |
| env = _sessions[req.session_id]["env"] | |
| action = Action( | |
| action_type=req.action_type, | |
| item=req.item, | |
| x=req.x, | |
| y=req.y, | |
| rotation=req.rotation | |
| ) | |
| obs, reward, done, info = env.step(action) | |
| return { | |
| "observation": obs.model_dump(), | |
| "reward": reward, | |
| "done": done, | |
| "result": env.get_episode_result().model_dump() if done else None | |
| } | |
| def main(): | |
| import uvicorn | |
| uvicorn.run("server.app:app", host="0.0.0.0", port=7860, reload=True) | |
| if __name__ == "__main__": | |
| main() | |