Spaces:
Sleeping
Sleeping
File size: 1,853 Bytes
a33aae2 | 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 | """
OpenEnv Compatibility Shim
===========================
Minimal Environment ABC that mirrors the openenv-core interface.
Used for local dev on Python 3.9. In Docker/Colab (Python 3.10+),
the real openenv-core takes over automatically.
"""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any
class Environment(ABC):
"""Minimal OpenEnv Environment base class."""
@abstractmethod
def reset(self, **kwargs) -> Any:
...
@abstractmethod
def step(self, action: Any, **kwargs) -> Any:
...
@abstractmethod
def state(self) -> Any:
...
def create_app(env_factory, action_type, observation_type, max_concurrent_envs=1):
"""Create a FastAPI app wrapping the environment."""
from fastapi import FastAPI
import json
app = FastAPI(title="SynthAudit.Env")
envs = {}
@app.get("/health")
async def health():
return {"status": "ok"}
@app.post("/reset")
async def reset_env(body: dict = {}):
env = env_factory()
eid = id(env)
envs[eid] = env
obs = env.reset(**body)
return {"env_id": eid, "observation": obs.dict() if hasattr(obs, 'dict') else obs.model_dump()}
@app.post("/step/{env_id}")
async def step_env(env_id: int, action: dict):
env = envs.get(env_id)
if not env:
return {"error": "env not found"}
act = action_type(**action)
obs = env.step(act)
return {"observation": obs.dict() if hasattr(obs, 'dict') else obs.model_dump()}
@app.get("/state/{env_id}")
async def get_state(env_id: int):
env = envs.get(env_id)
if not env:
return {"error": "env not found"}
s = env.state()
return {"state": s.dict() if hasattr(s, 'dict') else s.model_dump()}
return app
|