| import os |
| import requests |
| import time |
| from typing import Any, Dict |
|
|
| class RevOpsEnvClient: |
| def __init__(self, base_url: str = "http://localhost:8000"): |
| self.base_url = base_url |
| self.max_retries = 3 |
|
|
| def _request(self, method: str, endpoint: str, json: Dict = None) -> Dict: |
| url = f"{self.base_url}{endpoint}" |
| for attempt in range(self.max_retries): |
| try: |
| response = requests.request(method, url, json=json) |
| response.raise_for_status() |
| return response.json() |
| except requests.exceptions.RequestException as e: |
| time.sleep(1) |
| raise Exception(f"Failed to reach env after {self.max_retries} attempts.") |
| |
| def reset(self, task_id: str) -> Dict[str, Any]: |
| return self._request("POST", f"/reset?episode_id={task_id}&task_id={task_id}", json={"episode_id": task_id, "task_id": task_id}) |
|
|
| def step(self, action: Dict[str, Any]) -> Dict[str, Any]: |
| return self._request("POST", "/step", json={"action": action}) |
|
|
| def state(self) -> Dict[str, Any]: |
| return self._request("GET", "/state") |
|
|
| def grader(self) -> float: |
| res = self._request("GET", "/grader") |
| return res.get("grader_score", 0.0) |
|
|
