Spaces:
Sleeping
Sleeping
File size: 1,163 Bytes
84607b3 | 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 | from __future__ import annotations
import threading
from typing import Any
from env import app as env_app
class InProcessEnvBackend:
def __init__(self) -> None:
self._lock = threading.Lock()
def call(self, endpoint: str, payload: dict[str, Any] | None = None) -> dict[str, Any]:
payload = payload or {}
with self._lock:
if endpoint == "reset":
return env_app.reset(payload)
if endpoint == "step":
return env_app.step(payload)
if endpoint == "state":
return env_app.get_state()
if endpoint == "health":
return env_app.health()
raise ValueError(f"Unsupported endpoint: {endpoint}")
def reset(self, payload: dict[str, Any] | None = None) -> dict[str, Any]:
return self.call("reset", payload)
def step(self, payload: dict[str, Any] | None = None) -> dict[str, Any]:
return self.call("step", payload)
def state(self) -> dict[str, Any]:
return self.call("state")
def health(self) -> dict[str, Any]:
return self.call("health")
BACKEND = InProcessEnvBackend()
|