Spaces:
Sleeping
Sleeping
File size: 1,700 Bytes
38d40b2 | 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 | from __future__ import annotations
from typing import Any
import requests
from models import (
AgenticTrafficAction,
AgenticTrafficObservation,
AgenticTrafficState,
)
class AgenticTrafficClient:
"""Thin HTTP client for the DistrictFlow OpenEnv server."""
def __init__(self, base_url: str):
self.base_url = base_url.rstrip("/")
def reset(self, seed: int | None = None) -> AgenticTrafficObservation:
response = requests.post(
f"{self.base_url}/reset",
json={"seed": seed},
timeout=60,
)
response.raise_for_status()
payload = response.json()
return AgenticTrafficObservation.model_validate(payload["observation"])
def step(self, action: AgenticTrafficAction) -> AgenticTrafficObservation:
response = requests.post(
f"{self.base_url}/step",
json={"action": action.model_dump()},
timeout=60,
)
response.raise_for_status()
payload = response.json()
observation = AgenticTrafficObservation.model_validate(payload["observation"])
observation.done = bool(payload.get("done", False))
observation.reward = float(payload.get("reward", 0.0))
return observation
def state(self) -> AgenticTrafficState:
response = requests.get(f"{self.base_url}/state", timeout=60)
response.raise_for_status()
payload = response.json()
return AgenticTrafficState.model_validate(payload["state"])
def health(self) -> dict[str, Any]:
response = requests.get(f"{self.base_url}/health", timeout=30)
response.raise_for_status()
return response.json()
|