examples.example_01.world

Raw Gymnasium wrapper conforming to the OaK World protocol.

This wrapper intentionally returns the original environment observations. Discovery and perception are responsible for turning those raw values into the normalized AgentObservation and structured subjective state.

 1"""Raw Gymnasium wrapper conforming to the OaK World protocol.
 2
 3This wrapper intentionally returns the original environment observations.
 4Discovery and perception are responsible for turning those raw values into
 5the normalized `AgentObservation` and structured subjective state.
 6"""
 7
 8from __future__ import annotations
 9
10from collections.abc import Mapping
11from typing import Any, Generic, TypeVar, cast
12
13import gymnasium as gym
14
15from oak.interfaces import World
16from oak.types import TimeStep
17
18
19GymObsT = TypeVar("GymObsT")
20GymActionT = TypeVar("GymActionT")
21
22
23class GymWorld(World[GymObsT, GymActionT, dict[str, Any]], Generic[GymObsT, GymActionT]):
24    """Thin Gymnasium wrapper with no embedded observation semantics."""
25
26    def __init__(
27        self,
28        env_id: str,
29        *,
30        make_kwargs: Mapping[str, Any] | None = None,
31    ) -> None:
32        self.env_id = env_id
33        self.env: gym.Env[GymObsT, GymActionT] = cast(
34            gym.Env[GymObsT, GymActionT],
35            gym.make(env_id, **dict(make_kwargs or {})),
36        )
37
38    def reset(self) -> TimeStep[GymObsT, dict[str, Any]]:
39        obs, info = self.env.reset()
40        return TimeStep(observation=obs, reward=0.0, info=info)
41
42    def step(self, action: GymActionT) -> TimeStep[GymObsT, dict[str, Any]]:
43        obs, reward, terminated, truncated, info = self.env.step(action)
44        return TimeStep(
45            observation=obs,
46            reward=float(reward),
47            terminated=terminated,
48            truncated=truncated,
49            info=info,
50        )
51
52    def render_frame(self) -> Any | None:
53        """Return the current rendered frame when the env was created with rendering."""
54        try:
55            return self.env.render()
56        except Exception:
57            return None
58
59    def close(self) -> None:
60        self.env.close()
class GymWorld(oak.interfaces.World[~GymObsT, ~GymActionT, dict[str, typing.Any]], typing.Generic[~GymObsT, ~GymActionT]): 
24class GymWorld(World[GymObsT, GymActionT, dict[str, Any]], Generic[GymObsT, GymActionT]):
25    """Thin Gymnasium wrapper with no embedded observation semantics."""
26
27    def __init__(
28        self,
29        env_id: str,
30        *,
31        make_kwargs: Mapping[str, Any] | None = None,
32    ) -> None:
33        self.env_id = env_id
34        self.env: gym.Env[GymObsT, GymActionT] = cast(
35            gym.Env[GymObsT, GymActionT],
36            gym.make(env_id, **dict(make_kwargs or {})),
37        )
38
39    def reset(self) -> TimeStep[GymObsT, dict[str, Any]]:
40        obs, info = self.env.reset()
41        return TimeStep(observation=obs, reward=0.0, info=info)
42
43    def step(self, action: GymActionT) -> TimeStep[GymObsT, dict[str, Any]]:
44        obs, reward, terminated, truncated, info = self.env.step(action)
45        return TimeStep(
46            observation=obs,
47            reward=float(reward),
48            terminated=terminated,
49            truncated=truncated,
50            info=info,
51        )
52
53    def render_frame(self) -> Any | None:
54        """Return the current rendered frame when the env was created with rendering."""
55        try:
56            return self.env.render()
57        except Exception:
58            return None
59
60    def close(self) -> None:
61        self.env.close()

Thin Gymnasium wrapper with no embedded observation semantics.

GymWorld(env_id: 'str', *, make_kwargs: 'Mapping[str, Any] | None' = None) 
27    def __init__(
28        self,
29        env_id: str,
30        *,
31        make_kwargs: Mapping[str, Any] | None = None,
32    ) -> None:
33        self.env_id = env_id
34        self.env: gym.Env[GymObsT, GymActionT] = cast(
35            gym.Env[GymObsT, GymActionT],
36            gym.make(env_id, **dict(make_kwargs or {})),
37        )
env_id
env: 'gym.Env[GymObsT, GymActionT]'
def reset(self) -> 'TimeStep[GymObsT, dict[str, Any]]': 
39    def reset(self) -> TimeStep[GymObsT, dict[str, Any]]:
40        obs, info = self.env.reset()
41        return TimeStep(observation=obs, reward=0.0, info=info)
def step(self, action: 'GymActionT') -> 'TimeStep[GymObsT, dict[str, Any]]': 
43    def step(self, action: GymActionT) -> TimeStep[GymObsT, dict[str, Any]]:
44        obs, reward, terminated, truncated, info = self.env.step(action)
45        return TimeStep(
46            observation=obs,
47            reward=float(reward),
48            terminated=terminated,
49            truncated=truncated,
50            info=info,
51        )
def render_frame(self) -> 'Any | None': 
53    def render_frame(self) -> Any | None:
54        """Return the current rendered frame when the env was created with rendering."""
55        try:
56            return self.env.render()
57        except Exception:
58            return None

Return the current rendered frame when the env was created with rendering.

def close(self) -> 'None': 
60    def close(self) -> None:
61        self.env.close()

Release environment resources. Default is a no-op.