Spaces:
Sleeping
Sleeping
File size: 1,686 Bytes
3da7d69 6fb2c0c 3da7d69 6fb2c0c 3da7d69 e7de479 | 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 | """
FastAPI app exposing the warehouse fulfillment environment.
"""
from __future__ import annotations
from typing import Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from grid_env.Server.warehouse_env import WarehouseEnvService
app = FastAPI(
title="Warehouse Fulfillment Env Server",
version="0.1.0",
description="HTTP server for the MiniGrid-style warehouse fulfillment environment.",
)
service = WarehouseEnvService()
class ResetRequest(BaseModel):
task_id: Optional[str] = Field(default=None, description="Task ID to load on reset.")
seed: Optional[int] = Field(default=None, description="Optional deterministic seed.")
class StepRequest(BaseModel):
command: str = Field(description="Environment action command.")
@app.get("/health")
def health() -> dict:
return service.health()
@app.get("/tasks")
def tasks() -> dict:
return service.tasks()
@app.post("/reset")
def reset(payload: ResetRequest = None) -> dict:
task_id = payload.task_id if payload else None
seed = payload.seed if payload else None
try:
return service.reset(task_id=task_id, seed=seed)
except KeyError as exc:
raise HTTPException(status_code=404, detail=str(exc)) from exc
@app.post("/step")
def step(payload: StepRequest) -> dict:
try:
return service.step(command=payload.command)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@app.get("/state")
def state() -> dict:
return service.state()
def main() -> None:
import uvicorn
uvicorn.run("grid_env.Server.app:app", host="0.0.0.0", port=8000, reload=False)
|