Spaces:
Running
Running
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| from typing import Any, Dict | |
| import requests | |
| class GDPR_EraseOps_Client: | |
| """Lightweight HTTP client for the GDPR EraseOps environment.""" | |
| base_url: str = "http://localhost:7860" | |
| timeout: int = 20 | |
| def reset(self, task_name: str = "clean_sweep", seed: int = 42) -> Dict[str, Any]: | |
| response = requests.post( | |
| f"{self.base_url}/reset", | |
| params={"task_name": task_name, "seed": seed}, | |
| timeout=self.timeout, | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |
| def step(self, action: Dict[str, Any]) -> Dict[str, Any]: | |
| response = requests.post(f"{self.base_url}/step", json=action, timeout=self.timeout) | |
| response.raise_for_status() | |
| return response.json() | |
| def state(self) -> Dict[str, Any]: | |
| response = requests.get(f"{self.base_url}/state", timeout=self.timeout) | |
| response.raise_for_status() | |
| return response.json() | |
| def tasks(self) -> Dict[str, Any]: | |
| response = requests.get(f"{self.base_url}/tasks", timeout=self.timeout) | |
| response.raise_for_status() | |
| return response.json() | |
| def grade(self, task_name: str) -> Dict[str, Any]: | |
| response = requests.post( | |
| f"{self.base_url}/grade", | |
| params={"task_name": task_name}, | |
| timeout=self.timeout, | |
| ) | |
| response.raise_for_status() | |
| return response.json() | |