Spaces:
Sleeping
Sleeping
| """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), | |
| ) | |