Spaces:
Sleeping
Sleeping
File size: 1,882 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 55 56 57 58 59 60 61 62 63 | """Compatibility helpers for local development without openenv-core."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Generic, TypeVar
from pydantic import BaseModel, Field
ActionT = TypeVar("ActionT")
ObservationT = TypeVar("ObservationT")
StateT = TypeVar("StateT")
try: # pragma: no cover - exercised when openenv-core is installed
from openenv.core import EnvClient
from openenv.core.client_types import StepResult
from openenv.core.env_server.interfaces import Environment
from openenv.core.env_server.types import Action, Observation, State
OPENENV_AVAILABLE = True
except ImportError: # pragma: no cover - local fallback for unit tests
OPENENV_AVAILABLE = False
class Action(BaseModel):
"""Fallback action model."""
class Observation(BaseModel):
"""Fallback observation model."""
reward: float | None = None
done: bool = False
metadata: dict[str, Any] = Field(default_factory=dict)
class State(BaseModel):
"""Fallback transport state."""
episode_id: str | None = None
step_count: int = 0
class Environment(Generic[ObservationT, ActionT]):
"""Fallback environment interface."""
SUPPORTS_CONCURRENT_SESSIONS: bool = False
@dataclass
class StepResult(Generic[ObservationT]):
"""Fallback step result container."""
observation: ObservationT
reward: float | None = None
done: bool = False
class EnvClient(Generic[ActionT, ObservationT, StateT]):
"""Fallback client base class used only for local parsing tests."""
def __init__(self, *args: Any, **kwargs: Any) -> None:
self._base_url = kwargs.get("base_url")
def close(self) -> None:
"""Mirror the real client close method."""
return None
|