Spaces:
Sleeping
Sleeping
File size: 3,818 Bytes
6474552 3845661 | 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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """
server/app.py – FastAPI server for Smart Farm Resource Manager.
Exposes:
POST /reset -> Observation
POST /step -> StepResult
GET /state -> dict
GET /health -> dict
GET /tasks -> dict
"""
from __future__ import annotations
import json
import os
from typing import Any, Dict
import uvicorn
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from src.envs.smart_farm_env.models import Action, Observation, StepResult, TaskConfig
from src.envs.smart_farm_env.server.environment import SmartFarmEnv
# ---------------------------------------------------------------------------
# App
# ---------------------------------------------------------------------------
app = FastAPI(
title="Smart Farm Resource Manager",
description="OpenEnv-compatible precision agriculture simulation API",
version="1.0.0",
docs_url="/docs", # default, but explicit
redoc_url="/redoc",
)
@app.get("/")
async def root():
return {
"message": "🚜 Smart Farm Resource Manager API is running!",
"version": "1.0.0",
"docs": "/docs",
"redoc": "/redoc",
"available_endpoints": [
"/health",
"/reset",
"/step",
"/state",
"/tasks"
]
}
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# Singleton env (one active episode per server instance)
_env: SmartFarmEnv | None = None
# Load task configs once at startup
_TASKS_PATH = os.path.join(os.path.dirname(__file__), "..", "tasks", "tasks.json")
with open(_TASKS_PATH) as _f:
_tasks_raw = json.load(_f)["tasks"]
TASK_MAP: Dict[str, TaskConfig] = {t["name"]: TaskConfig(**t) for t in _tasks_raw}
def _require_env() -> SmartFarmEnv:
if _env is None:
raise HTTPException(status_code=400, detail="Not initialised — call POST /reset first.")
return _env
# ---------------------------------------------------------------------------
# Endpoints
# ---------------------------------------------------------------------------
@app.get("/health")
def health() -> Dict[str, str]:
return {"status": "ok", "env": "smart-farm-env"}
@app.post("/reset", response_model=Observation)
def reset(task_name: str = "easy") -> Observation:
global _env
if task_name not in TASK_MAP:
raise HTTPException(
status_code=404,
detail=f"Unknown task '{task_name}'. Available: {list(TASK_MAP)}",
)
_env = SmartFarmEnv(TASK_MAP[task_name])
return _env.reset()
@app.post("/step", response_model=StepResult)
def step(action: Action) -> StepResult:
env = _require_env()
try:
return env.step(action)
except RuntimeError as exc:
raise HTTPException(status_code=400, detail=str(exc))
@app.get("/state")
def state() -> Dict[str, Any]:
return _require_env().state()
@app.get("/tasks")
def list_tasks() -> Dict[str, Any]:
return {
"tasks": [
{
"name": t.name,
"description": t.description,
"num_plots": t.num_plots,
"allowed_actions": t.allowed_actions,
}
for t in TASK_MAP.values()
]
}
# ---------------------------------------------------------------------------
# Entry point (used by server Dockerfile CMD)
# ---------------------------------------------------------------------------
if __name__ == "__main__":
port = int(os.environ.get("PORT", 7860))
uvicorn.run("src.envs.smart_farm_env.server.app:app", host="0.0.0.0", port=port, reload=False)
def main():
import uvicorn
uvicorn.run(app, host='0.0.0.0', port=7860)
if __name__ == '__main__':
main()
|