Spaces:
Sleeping
Sleeping
File size: 3,639 Bytes
e3e3da2 b4fec30 ab38db6 e3e3da2 ab38db6 6b8c880 0dd5d30 e3e3da2 a95dc70 6b8c880 e3e3da2 ab38db6 b4fec30 ab38db6 b4fec30 e3e3da2 a95dc70 e3e3da2 a95dc70 e3e3da2 5f4d425 a95dc70 e3e3da2 2ddf5f5 e3e3da2 262e2fd e3e3da2 | 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 | from fastapi import FastAPI
from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from pydantic import BaseModel
from typing import Optional
import os
from netzero_nav.env import NetZeroEnv
from netzero_nav.models import Action, Observation, StepResponse
app = FastAPI(title="NetZero Nav: Eco-Resilient Logistics", version="1.0.0")
_env = NetZeroEnv()
# Mount Static Dashboard
static_path = os.path.join(os.getcwd(), "dashboard")
app.mount("/dashboard", StaticFiles(directory=static_path), name="dashboard")
@app.get("/")
def root():
return RedirectResponse(url="/dashboard/index.html")
class ResetRequest(BaseModel):
task: Optional[str] = "easy"
seed: Optional[int] = 42
# --- OpenEnv Required Endpoints ---
@app.get("/health")
def health():
return {"status": "healthy"}
@app.get("/metadata")
def metadata():
return {
"name": "netzero-nav",
"description": "Autonomous RL logistics agent navigating global disruptions with a Net-Zero carbon mandate.",
"version": "1.0.0",
"tasks": ["easy", "medium", "hard"],
"tags": ["logistics", "esg", "supply-chain", "rl"]
}
@app.get("/schema")
def schema():
return {
"action": {
"action_type": {"type": "string", "enum": ["order_parts", "produce", "offset", "skip", "cancel"]},
"part_type": {"type": "string", "optional": True},
"quantity": {"type": "integer", "optional": True},
"mode": {"type": "string", "enum": ["sea", "air", "rail", "road"], "optional": True},
"product": {"type": "string", "optional": True},
"offset_amount": {"type": "number", "optional": True},
"shipment_id": {"type": "string", "optional": True}
},
"observation": {
"step": {"type": "integer"},
"inventory": {"type": "object"},
"active_shipments": {"type": "array"},
"pending_orders": {"type": "array"},
"carbon_total": {"type": "number"},
"carbon_quota": {"type": "number"},
"cash_balance": {"type": "number"},
"news": {"type": "string", "optional": True}
},
"state": {
"step": {"type": "integer"},
"inventory": {"type": "object"},
"carbon_total": {"type": "number"},
"cash_balance": {"type": "number"},
"orders": {"type": "array"},
"active_shipments": {"type": "array"}
}
}
@app.post("/mcp")
def mcp(body: dict = None):
return {
"jsonrpc": "2.0",
"id": (body or {}).get("id"),
"result": {
"tools": ["reset", "step", "state", "metadata", "schema"],
"env": "netzero-nav"
}
}
# --- Core Env Endpoints ---
@app.post("/reset", response_model=Observation)
def reset(req: ResetRequest = None):
if req is None:
req = ResetRequest()
_env.task = req.task or "easy"
return _env.reset(seed=req.seed)
@app.post("/step", response_model=StepResponse)
def step(action: Action):
obs, reward, done, info = _env.step(action)
return StepResponse(observation=obs, reward=reward, done=done, info=info)
@app.post("/trigger")
def trigger():
_env.sea_blocked_until = _env.step_count + 7
return {"status": "Suez Jam Triggered", "duration": 7}
@app.get("/state")
def state():
return {
"step": _env.step_count,
"inventory": _env.inventory,
"carbon": _env.carbon_total,
"cash": _env.cash_balance,
"orders": _env.pending_orders,
"active_shipments": _env.active_shipments
}
|