Spaces:
Sleeping
Sleeping
File size: 2,594 Bytes
cacd58c f60ce5a cacd58c | 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 | # server/app.py
from fastapi import FastAPI, HTTPException
from openenv.core.env_server import create_fastapi_app
from ..models import Action, Observation, TaskInfo
from .environment import CodeDebugEnvironment
from .tasks import TASK_REGISTRY
from .grader import grade
# Core OpenEnv app (provides /reset, /step, /state, /ws, /health)
app = create_fastapi_app(CodeDebugEnvironment, Action, Observation)
# ── Additional required hackathon endpoints ────────────────────────────
@app.get("/")
def home():
return {
"message": "Welcome to the OpenEnv Code Debug Environment!",
"version": "1.0.0",
"endpoints": ["/health", "/tasks", "/reset", "/step", "/grader", "/baseline"],
"docs": "https://huggingface.co/spaces/luciferai-devil/code-debug-env"
}
@app.get("/tasks")
def list_tasks() -> list[TaskInfo]:
"""Return all tasks with their action schema."""
return [
TaskInfo(
task_id=tid,
difficulty=task["difficulty"],
description=task["description"],
action_schema=Action.model_json_schema(),
)
for tid, task in TASK_REGISTRY.items()
]
@app.get("/grader")
def get_grader_score(task_id: str, submitted_code: str) -> dict:
"""
Grade a submission directly (for testing / evaluation).
Returns: { score: float, passed: int, total: int, test_results: list }
"""
if task_id not in TASK_REGISTRY:
raise HTTPException(status_code=404, detail=f"Unknown task_id: {task_id}")
task = TASK_REGISTRY[task_id]
result = grade(submitted_code, task_id, task["test_suite"])
return {
"task_id": task_id,
"score": result["score"],
"passed": result["passed"],
"total": result["total"],
"test_results": [r.model_dump() for r in result["test_results"]],
}
@app.get("/baseline")
def run_baseline() -> dict:
"""
Run the baseline agent on all tasks and return scores.
This endpoint triggers the baseline inference script.
"""
import subprocess, sys, json
try:
result = subprocess.run(
[sys.executable, "baseline/run_baseline.py", "--output", "json"],
capture_output=True, text=True, timeout=120,
)
return json.loads(result.stdout)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
def main():
"""Entry point for the server."""
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
if __name__ == "__main__":
main()
|