Spaces:
Sleeping
Sleeping
File size: 3,198 Bytes
0085ac8 17a5579 0085ac8 17a5579 0085ac8 17a5579 125f529 17a5579 0085ac8 17a5579 0085ac8 | 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 | 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")
@app.get("/")
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 = {}
@app.get("/api/v1/tasks")
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 ──────────────────────────────────────
@app.post("/reset")
async def reset():
"""Required by OpenEnv pre-validation script (validate-submission.sh)."""
return {"status": "ok", "tasks": list(TASKS.keys())}
@app.get("/health")
async def health():
"""Docker HEALTHCHECK support."""
return {"status": "healthy", "tasks": len(TASKS)}
@app.post("/api/v1/session/start")
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()}
@app.post("/api/v1/session/step")
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()
|