Spaces:
Sleeping
Sleeping
File size: 1,826 Bytes
b7a5bab | 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 | """Client for the Store OpenEnv simulation."""
from __future__ import annotations
from typing import Any
try:
from ._compat import EnvClient, State, StepResult
from .models import StoreAction, StoreObservation, dump_model
except ImportError: # pragma: no cover - supports direct module imports
from _compat import EnvClient, State, StepResult
from models import StoreAction, StoreObservation, dump_model
class StoreEnv(
EnvClient[StoreAction, StoreObservation, State]
):
"""Client wrapper for the store environment."""
def _step_payload(self, action: StoreAction) -> dict[str, Any]:
"""Convert a store action into a JSON payload."""
return dump_model(action, exclude_none=True, exclude_defaults=True)
def _parse_result(self, payload: dict[str, Any]) -> StepResult[StoreObservation]:
"""Parse a server response into a typed observation."""
obs_data = payload.get("observation", {})
clean_obs_data = {
key: value
for key, value in obs_data.items()
if key not in {"done", "reward", "metadata"}
}
observation = StoreObservation(
**clean_obs_data,
done=payload.get("done", obs_data.get("done", False)),
reward=payload.get("reward", obs_data.get("reward")),
metadata=obs_data.get("metadata", payload.get("metadata", {})),
)
return StepResult(
observation=observation,
reward=payload.get("reward"),
done=payload.get("done", False),
)
def _parse_state(self, payload: dict[str, Any]) -> State:
"""Parse the OpenEnv transport state payload."""
return State(
episode_id=payload.get("episode_id"),
step_count=payload.get("step_count", 0),
)
|