Spaces:
Runtime error
Runtime error
File size: 2,713 Bytes
205f6c7 01468da 205f6c7 01468da 205f6c7 01468da 205f6c7 01468da 5d68774 01468da 5d68774 01468da 5d68774 01468da 5d68774 01468da 5d68774 01468da 205f6c7 01468da 205f6c7 01468da 205f6c7 01468da 5d68774 01468da 205f6c7 01468da 5d68774 205f6c7 01468da 205f6c7 01468da 205f6c7 01468da 5d68774 01468da 205f6c7 01468da 205f6c7 01468da | 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 | """
HF Spaces server — DebugOps Environment (OpenEnv compatible)
Exposes:
POST /reset
POST /step
GET /state
"""
from __future__ import annotations
from typing import Dict, Any, Optional
import os
import uvicorn
from fastapi import FastAPI, HTTPException, Body
from tasks.task_simple import create_env as create_simple
from tasks.task_multi_service import create_env as create_multi
from tasks.task_critical import create_env as create_critical
app = FastAPI(title="DebugOps AI Environment", version="1.0.0")
# Global environment instance
_env: Optional[Any] = None
def create_env(task: str):
if task == "simple":
return create_simple()
elif task == "multi_service":
return create_multi()
elif task == "critical":
return create_critical()
else:
raise ValueError(f"Invalid task: {task}")
def get_env():
global _env
if _env is None:
raise HTTPException(status_code=400, detail="Call /reset first")
return _env
@app.get("/")
def root():
return {
"name": "DebugOps AI Environment",
"description": "Production debugging RL environment (OpenEnv compatible)",
"endpoints": ["/reset", "/step", "/state", "/health"],
}
@app.get("/health")
def health():
return {"status": "ok"}
@app.post("/reset")
def reset(payload: Optional[Dict[str, Any]] = Body(default={})):
"""
Accepts BOTH:
{} (validator case)
{"task": "critical"} (manual case)
"""
global _env
try:
task = payload.get("task", "simple") if payload else "simple"
_env = create_env(task)
obs = _env.reset()
return {
"observation": obs,
"done": False,
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/step")
@app.post("/step/")
def step(payload: Dict[str, Any] = Body(...)):
env = get_env()
try:
action = payload.get("action")
if not action:
raise ValueError("Missing 'action' field")
obs, reward, done, info = env.step(action)
return {
"observation": obs,
"reward": float(round(reward, 3)),
"done": done,
"info": info,
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/state")
@app.get("/state/")
def state():
env = get_env()
try:
return env.state()
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
def main():
port = int(os.getenv("PORT", 7860))
uvicorn.run(app, host="0.0.0.0", port=port)
if __name__ == "__main__":
main() |