We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Bug Report] gym normalize wrappers are incompatible with envpool
**Describe the bug**
`gym.wrappers.NormalizeObservation` and `gym.wrappers.NormalizeReward` are incompatible with envpool. See https://github.com/sail-sg/envpool/issues/185

**Code example**

```python
import numpy as np
import envpool
import gym

envs = envpool.make(
    "HalfCheetah-v4",
    env_type="gym",
    num_envs=4,
)
envs.num_envs = 4
envs.single_action_space = envs.action_space
envs.single_observation_space = envs.observation_space
envs.is_vector_env = True
envs = gym.wrappers.ClipAction(envs)
envs = gym.wrappers.NormalizeObservation(envs)
envs = gym.wrappers.TransformObservation(envs, lambda obs: np.clip(obs, -10, 10))
envs = gym.wrappers.NormalizeReward(envs)
envs = gym.wrappers.TransformReward(envs, lambda reward: np.clip(reward, -10, 10))
obs = envs.reset()
envs.step(np.array([envs.action_space.sample() for _ in range(envs.num_envs)]))
```
```
Traceback (most recent call last):
  File "/home/costa/Documents/go/src/github.com/vwxyzjn/envpool-cleanrl/bug.py", line 22, in <module>
    envs.step(np.array([envs.action_space.sample() for _ in range(envs.num_envs)]))
  File "/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/core.py", line 532, in step
    step_returns = self.env.step(action)
  File "/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/wrappers/normalize.py", line 149, in step
    self.env.step(action), True, self.is_vector_env
  File "/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/core.py", line 493, in step
    step_returns = self.env.step(action)
  File "/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/wrappers/normalize.py", line 77, in step
    obs, rews, terminateds, truncateds, infos = step_api_compatibility(
  File "/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/utils/step_api_compatibility.py", line 178, in step_api_compatibility
    return step_to_new_api(step_returns, is_vector_env)
  File "/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/utils/step_api_compatibility.py", line 59, in step_to_new_api
    and not infos["_TimeLimit.truncated"][i]
KeyError: '_TimeLimit.truncated'
```

**System Info**
Describe the characteristic of your environment:
 * Describe how Gym was installed (pip, docker, source, ...) pip
 * What OS/version of Linux you're using. Note that while we will accept PRs to improve Window's support, we do not officially support it. Linux.
 * Python version: 3.9

**Additional context**
Add any other context about the problem here.

### Checklist

- [x] I have checked that there is no similar [issue](https://github.com/openai/gym/issues) in the repo (**required**)


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `gym/utils/step_api_compatibility.py`
Content:
```
1 """Contains methods for step compatibility, from old-to-new and new-to-old API, to be removed in 1.0."""
2 from typing import Tuple, Union
3 
4 import numpy as np
5 
6 from gym.core import ObsType
7 
8 OldStepType = Tuple[
9     Union[ObsType, np.ndarray],
10     Union[float, np.ndarray],
11     Union[bool, np.ndarray],
12     Union[dict, list],
13 ]
14 
15 NewStepType = Tuple[
16     Union[ObsType, np.ndarray],
17     Union[float, np.ndarray],
18     Union[bool, np.ndarray],
19     Union[bool, np.ndarray],
20     Union[dict, list],
21 ]
22 
23 
24 def step_to_new_api(
25     step_returns: Union[OldStepType, NewStepType], is_vector_env=False
26 ) -> NewStepType:
27     """Function to transform step returns to new step API irrespective of input API.
28 
29     Args:
30         step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)
31         is_vector_env (bool): Whether the step_returns are from a vector environment
32     """
33     if len(step_returns) == 5:
34         return step_returns
35     else:
36         assert len(step_returns) == 4
37         observations, rewards, dones, infos = step_returns
38 
39         terminateds = []
40         truncateds = []
41         if not is_vector_env:
42             dones = [dones]
43 
44         for i in range(len(dones)):
45             # For every condition, handling - info single env /  info vector env (list) / info vector env (dict)
46 
47             # TimeLimit.truncated attribute not present - implies either terminated or episode still ongoing based on `done`
48             if (not is_vector_env and "TimeLimit.truncated" not in infos) or (
49                 is_vector_env
50                 and (
51                     (
52                         isinstance(infos, list)
53                         and "TimeLimit.truncated" not in infos[i]
54                     )  # vector env, list info api
55                     or (
56                         "TimeLimit.truncated" not in infos
57                         or (
58                             "TimeLimit.truncated" in infos
59                             and not infos["_TimeLimit.truncated"][i]
60                         )
61                     )  # vector env, dict info api, if mask is False, it's the same as TimeLimit.truncated attribute not being present for env 'i'
62                 )
63             ):
64 
65                 terminateds.append(dones[i])
66                 truncateds.append(False)
67 
68             # This means info["TimeLimit.truncated"] exists and this elif checks if it is True, which means the truncation has occurred but termination has not.
69             elif (
70                 infos["TimeLimit.truncated"]
71                 if not is_vector_env
72                 else (
73                     infos["TimeLimit.truncated"][i]
74                     if isinstance(infos, dict)
75                     else infos[i]["TimeLimit.truncated"]
76                 )
77             ):
78                 assert dones[i]
79                 terminateds.append(False)
80                 truncateds.append(True)
81             else:
82                 # This means info["TimeLimit.truncated"] exists but is False, which means the core environment had already terminated,
83                 # but it also exceeded maximum timesteps at the same step.
84                 assert dones[i]
85                 terminateds.append(True)
86                 truncateds.append(True)
87 
88         return (
89             observations,
90             rewards,
91             np.array(terminateds, dtype=np.bool_) if is_vector_env else terminateds[0],
92             np.array(truncateds, dtype=np.bool_) if is_vector_env else truncateds[0],
93             infos,
94         )
95 
96 
97 def step_to_old_api(
98     step_returns: Union[NewStepType, OldStepType], is_vector_env: bool = False
99 ) -> OldStepType:
100     """Function to transform step returns to old step API irrespective of input API.
101 
102     Args:
103         step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)
104         is_vector_env (bool): Whether the step_returns are from a vector environment
105     """
106     if len(step_returns) == 4:
107         return step_returns
108     else:
109         assert len(step_returns) == 5
110         observations, rewards, terminateds, truncateds, infos = step_returns
111         dones = []
112         if not is_vector_env:
113             terminateds = [terminateds]
114             truncateds = [truncateds]
115 
116         n_envs = len(terminateds)
117 
118         for i in range(n_envs):
119             dones.append(terminateds[i] or truncateds[i])
120             if truncateds[i]:
121                 if is_vector_env:
122                     # handle vector infos for dict and list
123                     if isinstance(infos, dict):
124                         if "TimeLimit.truncated" not in infos:
125                             # TODO: This should ideally not be done manually and should use vector_env's _add_info()
126                             infos["TimeLimit.truncated"] = np.zeros(n_envs, dtype=bool)
127                             infos["_TimeLimit.truncated"] = np.zeros(n_envs, dtype=bool)
128 
129                         infos["TimeLimit.truncated"][i] = (
130                             not terminateds[i] or infos["TimeLimit.truncated"][i]
131                         )
132                         infos["_TimeLimit.truncated"][i] = True
133                     else:
134                         # if vector info is a list
135                         infos[i]["TimeLimit.truncated"] = not terminateds[i] or infos[
136                             i
137                         ].get("TimeLimit.truncated", False)
138                 else:
139                     infos["TimeLimit.truncated"] = not terminateds[i] or infos.get(
140                         "TimeLimit.truncated", False
141                     )
142         return (
143             observations,
144             rewards,
145             np.array(dones, dtype=np.bool_) if is_vector_env else dones[0],
146             infos,
147         )
148 
149 
150 def step_api_compatibility(
151     step_returns: Union[NewStepType, OldStepType],
152     new_step_api: bool = False,
153     is_vector_env: bool = False,
154 ) -> Union[NewStepType, OldStepType]:
155     """Function to transform step returns to the API specified by `new_step_api` bool.
156 
157     Old step API refers to step() method returning (observation, reward, done, info)
158     New step API refers to step() method returning (observation, reward, terminated, truncated, info)
159     (Refer to docs for details on the API change)
160 
161     Args:
162         step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)
163         new_step_api (bool): Whether the output should be in new step API or old (False by default)
164         is_vector_env (bool): Whether the step_returns are from a vector environment
165 
166     Returns:
167         step_returns (tuple): Depending on `new_step_api` bool, it can return (obs, rew, done, info) or (obs, rew, terminated, truncated, info)
168 
169     Examples:
170         This function can be used to ensure compatibility in step interfaces with conflicting API. Eg. if env is written in old API,
171          wrapper is written in new API, and the final step output is desired to be in old API.
172 
173         >>> obs, rew, done, info = step_api_compatibility(env.step(action))
174         >>> obs, rew, terminated, truncated, info = step_api_compatibility(env.step(action), new_step_api=True)
175         >>> observations, rewards, dones, infos = step_api_compatibility(vec_env.step(action), is_vector_env=True)
176     """
177     if new_step_api:
178         return step_to_new_api(step_returns, is_vector_env)
179     else:
180         return step_to_old_api(step_returns, is_vector_env)
181
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/gym/utils/step_api_compatibility.py b/gym/utils/step_api_compatibility.py
--- a/gym/utils/step_api_compatibility.py
+++ b/gym/utils/step_api_compatibility.py
@@ -56,12 +56,15 @@
                         "TimeLimit.truncated" not in infos
                         or (
                             "TimeLimit.truncated" in infos
-                            and not infos["_TimeLimit.truncated"][i]
+                            and not infos["TimeLimit.truncated"][i]
                         )
-                    )  # vector env, dict info api, if mask is False, it's the same as TimeLimit.truncated attribute not being present for env 'i'
+                    )
+                    # vector env, dict info api, for env i, vector mask `_TimeLimit.truncated` is not considered, to be compatible with envpool
+                    # For env i, `TimeLimit.truncated` not being present is treated same as being present and set to False.
+                    # therefore, terminated=True, truncated=True simultaneously is not allowed while using compatibility functions
+                    # with vector info
                 )
             ):
-
                 terminateds.append(dones[i])
                 truncateds.append(False)
 
@@ -80,10 +83,11 @@
                 truncateds.append(True)
             else:
                 # This means info["TimeLimit.truncated"] exists but is False, which means the core environment had already terminated,
-                # but it also exceeded maximum timesteps at the same step.
+                # but it also exceeded maximum timesteps at the same step. However to be compatible with envpool, and to be backward compatible
+                # truncated is set to False here.
                 assert dones[i]
                 terminateds.append(True)
-                truncateds.append(True)
+                truncateds.append(False)
 
         return (
             observations,

{"golden_diff": "diff --git a/gym/utils/step_api_compatibility.py b/gym/utils/step_api_compatibility.py\n--- a/gym/utils/step_api_compatibility.py\n+++ b/gym/utils/step_api_compatibility.py\n@@ -56,12 +56,15 @@\n                         \"TimeLimit.truncated\" not in infos\n                         or (\n                             \"TimeLimit.truncated\" in infos\n-                            and not infos[\"_TimeLimit.truncated\"][i]\n+                            and not infos[\"TimeLimit.truncated\"][i]\n                         )\n-                    )  # vector env, dict info api, if mask is False, it's the same as TimeLimit.truncated attribute not being present for env 'i'\n+                    )\n+                    # vector env, dict info api, for env i, vector mask `_TimeLimit.truncated` is not considered, to be compatible with envpool\n+                    # For env i, `TimeLimit.truncated` not being present is treated same as being present and set to False.\n+                    # therefore, terminated=True, truncated=True simultaneously is not allowed while using compatibility functions\n+                    # with vector info\n                 )\n             ):\n-\n                 terminateds.append(dones[i])\n                 truncateds.append(False)\n \n@@ -80,10 +83,11 @@\n                 truncateds.append(True)\n             else:\n                 # This means info[\"TimeLimit.truncated\"] exists but is False, which means the core environment had already terminated,\n-                # but it also exceeded maximum timesteps at the same step.\n+                # but it also exceeded maximum timesteps at the same step. However to be compatible with envpool, and to be backward compatible\n+                # truncated is set to False here.\n                 assert dones[i]\n                 terminateds.append(True)\n-                truncateds.append(True)\n+                truncateds.append(False)\n \n         return (\n             observations,\n", "issue": "[Bug Report] gym normalize wrappers are incompatible with envpool\n**Describe the bug**\r\n`gym.wrappers.NormalizeObservation` and `gym.wrappers.NormalizeReward` are incompatible with envpool. See https://github.com/sail-sg/envpool/issues/185\r\n\r\n**Code example**\r\n\r\n```python\r\nimport numpy as np\r\nimport envpool\r\nimport gym\r\n\r\nenvs = envpool.make(\r\n    \"HalfCheetah-v4\",\r\n    env_type=\"gym\",\r\n    num_envs=4,\r\n)\r\nenvs.num_envs = 4\r\nenvs.single_action_space = envs.action_space\r\nenvs.single_observation_space = envs.observation_space\r\nenvs.is_vector_env = True\r\nenvs = gym.wrappers.ClipAction(envs)\r\nenvs = gym.wrappers.NormalizeObservation(envs)\r\nenvs = gym.wrappers.TransformObservation(envs, lambda obs: np.clip(obs, -10, 10))\r\nenvs = gym.wrappers.NormalizeReward(envs)\r\nenvs = gym.wrappers.TransformReward(envs, lambda reward: np.clip(reward, -10, 10))\r\nobs = envs.reset()\r\nenvs.step(np.array([envs.action_space.sample() for _ in range(envs.num_envs)]))\r\n```\r\n```\r\nTraceback (most recent call last):\r\n  File \"/home/costa/Documents/go/src/github.com/vwxyzjn/envpool-cleanrl/bug.py\", line 22, in <module>\r\n    envs.step(np.array([envs.action_space.sample() for _ in range(envs.num_envs)]))\r\n  File \"/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/core.py\", line 532, in step\r\n    step_returns = self.env.step(action)\r\n  File \"/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/wrappers/normalize.py\", line 149, in step\r\n    self.env.step(action), True, self.is_vector_env\r\n  File \"/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/core.py\", line 493, in step\r\n    step_returns = self.env.step(action)\r\n  File \"/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/wrappers/normalize.py\", line 77, in step\r\n    obs, rews, terminateds, truncateds, infos = step_api_compatibility(\r\n  File \"/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/utils/step_api_compatibility.py\", line 178, in step_api_compatibility\r\n    return step_to_new_api(step_returns, is_vector_env)\r\n  File \"/home/costa/.cache/pypoetry/virtualenvs/envpool-cleanrl-uAHoRI5J-py3.9/lib/python3.9/site-packages/gym/utils/step_api_compatibility.py\", line 59, in step_to_new_api\r\n    and not infos[\"_TimeLimit.truncated\"][i]\r\nKeyError: '_TimeLimit.truncated'\r\n```\r\n\r\n**System Info**\r\nDescribe the characteristic of your environment:\r\n * Describe how Gym was installed (pip, docker, source, ...) pip\r\n * What OS/version of Linux you're using. Note that while we will accept PRs to improve Window's support, we do not officially support it. Linux.\r\n * Python version: 3.9\r\n\r\n**Additional context**\r\nAdd any other context about the problem here.\r\n\r\n### Checklist\r\n\r\n- [x] I have checked that there is no similar [issue](https://github.com/openai/gym/issues) in the repo (**required**)\r\n\n", "before_files": [{"content": "\"\"\"Contains methods for step compatibility, from old-to-new and new-to-old API, to be removed in 1.0.\"\"\"\nfrom typing import Tuple, Union\n\nimport numpy as np\n\nfrom gym.core import ObsType\n\nOldStepType = Tuple[\n    Union[ObsType, np.ndarray],\n    Union[float, np.ndarray],\n    Union[bool, np.ndarray],\n    Union[dict, list],\n]\n\nNewStepType = Tuple[\n    Union[ObsType, np.ndarray],\n    Union[float, np.ndarray],\n    Union[bool, np.ndarray],\n    Union[bool, np.ndarray],\n    Union[dict, list],\n]\n\n\ndef step_to_new_api(\n    step_returns: Union[OldStepType, NewStepType], is_vector_env=False\n) -> NewStepType:\n    \"\"\"Function to transform step returns to new step API irrespective of input API.\n\n    Args:\n        step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n        is_vector_env (bool): Whether the step_returns are from a vector environment\n    \"\"\"\n    if len(step_returns) == 5:\n        return step_returns\n    else:\n        assert len(step_returns) == 4\n        observations, rewards, dones, infos = step_returns\n\n        terminateds = []\n        truncateds = []\n        if not is_vector_env:\n            dones = [dones]\n\n        for i in range(len(dones)):\n            # For every condition, handling - info single env /  info vector env (list) / info vector env (dict)\n\n            # TimeLimit.truncated attribute not present - implies either terminated or episode still ongoing based on `done`\n            if (not is_vector_env and \"TimeLimit.truncated\" not in infos) or (\n                is_vector_env\n                and (\n                    (\n                        isinstance(infos, list)\n                        and \"TimeLimit.truncated\" not in infos[i]\n                    )  # vector env, list info api\n                    or (\n                        \"TimeLimit.truncated\" not in infos\n                        or (\n                            \"TimeLimit.truncated\" in infos\n                            and not infos[\"_TimeLimit.truncated\"][i]\n                        )\n                    )  # vector env, dict info api, if mask is False, it's the same as TimeLimit.truncated attribute not being present for env 'i'\n                )\n            ):\n\n                terminateds.append(dones[i])\n                truncateds.append(False)\n\n            # This means info[\"TimeLimit.truncated\"] exists and this elif checks if it is True, which means the truncation has occurred but termination has not.\n            elif (\n                infos[\"TimeLimit.truncated\"]\n                if not is_vector_env\n                else (\n                    infos[\"TimeLimit.truncated\"][i]\n                    if isinstance(infos, dict)\n                    else infos[i][\"TimeLimit.truncated\"]\n                )\n            ):\n                assert dones[i]\n                terminateds.append(False)\n                truncateds.append(True)\n            else:\n                # This means info[\"TimeLimit.truncated\"] exists but is False, which means the core environment had already terminated,\n                # but it also exceeded maximum timesteps at the same step.\n                assert dones[i]\n                terminateds.append(True)\n                truncateds.append(True)\n\n        return (\n            observations,\n            rewards,\n            np.array(terminateds, dtype=np.bool_) if is_vector_env else terminateds[0],\n            np.array(truncateds, dtype=np.bool_) if is_vector_env else truncateds[0],\n            infos,\n        )\n\n\ndef step_to_old_api(\n    step_returns: Union[NewStepType, OldStepType], is_vector_env: bool = False\n) -> OldStepType:\n    \"\"\"Function to transform step returns to old step API irrespective of input API.\n\n    Args:\n        step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n        is_vector_env (bool): Whether the step_returns are from a vector environment\n    \"\"\"\n    if len(step_returns) == 4:\n        return step_returns\n    else:\n        assert len(step_returns) == 5\n        observations, rewards, terminateds, truncateds, infos = step_returns\n        dones = []\n        if not is_vector_env:\n            terminateds = [terminateds]\n            truncateds = [truncateds]\n\n        n_envs = len(terminateds)\n\n        for i in range(n_envs):\n            dones.append(terminateds[i] or truncateds[i])\n            if truncateds[i]:\n                if is_vector_env:\n                    # handle vector infos for dict and list\n                    if isinstance(infos, dict):\n                        if \"TimeLimit.truncated\" not in infos:\n                            # TODO: This should ideally not be done manually and should use vector_env's _add_info()\n                            infos[\"TimeLimit.truncated\"] = np.zeros(n_envs, dtype=bool)\n                            infos[\"_TimeLimit.truncated\"] = np.zeros(n_envs, dtype=bool)\n\n                        infos[\"TimeLimit.truncated\"][i] = (\n                            not terminateds[i] or infos[\"TimeLimit.truncated\"][i]\n                        )\n                        infos[\"_TimeLimit.truncated\"][i] = True\n                    else:\n                        # if vector info is a list\n                        infos[i][\"TimeLimit.truncated\"] = not terminateds[i] or infos[\n                            i\n                        ].get(\"TimeLimit.truncated\", False)\n                else:\n                    infos[\"TimeLimit.truncated\"] = not terminateds[i] or infos.get(\n                        \"TimeLimit.truncated\", False\n                    )\n        return (\n            observations,\n            rewards,\n            np.array(dones, dtype=np.bool_) if is_vector_env else dones[0],\n            infos,\n        )\n\n\ndef step_api_compatibility(\n    step_returns: Union[NewStepType, OldStepType],\n    new_step_api: bool = False,\n    is_vector_env: bool = False,\n) -> Union[NewStepType, OldStepType]:\n    \"\"\"Function to transform step returns to the API specified by `new_step_api` bool.\n\n    Old step API refers to step() method returning (observation, reward, done, info)\n    New step API refers to step() method returning (observation, reward, terminated, truncated, info)\n    (Refer to docs for details on the API change)\n\n    Args:\n        step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n        new_step_api (bool): Whether the output should be in new step API or old (False by default)\n        is_vector_env (bool): Whether the step_returns are from a vector environment\n\n    Returns:\n        step_returns (tuple): Depending on `new_step_api` bool, it can return (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n\n    Examples:\n        This function can be used to ensure compatibility in step interfaces with conflicting API. Eg. if env is written in old API,\n         wrapper is written in new API, and the final step output is desired to be in old API.\n\n        >>> obs, rew, done, info = step_api_compatibility(env.step(action))\n        >>> obs, rew, terminated, truncated, info = step_api_compatibility(env.step(action), new_step_api=True)\n        >>> observations, rewards, dones, infos = step_api_compatibility(vec_env.step(action), is_vector_env=True)\n    \"\"\"\n    if new_step_api:\n        return step_to_new_api(step_returns, is_vector_env)\n    else:\n        return step_to_old_api(step_returns, is_vector_env)\n", "path": "gym/utils/step_api_compatibility.py"}], "after_files": [{"content": "\"\"\"Contains methods for step compatibility, from old-to-new and new-to-old API, to be removed in 1.0.\"\"\"\nfrom typing import Tuple, Union\n\nimport numpy as np\n\nfrom gym.core import ObsType\n\nOldStepType = Tuple[\n    Union[ObsType, np.ndarray],\n    Union[float, np.ndarray],\n    Union[bool, np.ndarray],\n    Union[dict, list],\n]\n\nNewStepType = Tuple[\n    Union[ObsType, np.ndarray],\n    Union[float, np.ndarray],\n    Union[bool, np.ndarray],\n    Union[bool, np.ndarray],\n    Union[dict, list],\n]\n\n\ndef step_to_new_api(\n    step_returns: Union[OldStepType, NewStepType], is_vector_env=False\n) -> NewStepType:\n    \"\"\"Function to transform step returns to new step API irrespective of input API.\n\n    Args:\n        step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n        is_vector_env (bool): Whether the step_returns are from a vector environment\n    \"\"\"\n    if len(step_returns) == 5:\n        return step_returns\n    else:\n        assert len(step_returns) == 4\n        observations, rewards, dones, infos = step_returns\n\n        terminateds = []\n        truncateds = []\n        if not is_vector_env:\n            dones = [dones]\n\n        for i in range(len(dones)):\n            # For every condition, handling - info single env /  info vector env (list) / info vector env (dict)\n\n            # TimeLimit.truncated attribute not present - implies either terminated or episode still ongoing based on `done`\n            if (not is_vector_env and \"TimeLimit.truncated\" not in infos) or (\n                is_vector_env\n                and (\n                    (\n                        isinstance(infos, list)\n                        and \"TimeLimit.truncated\" not in infos[i]\n                    )  # vector env, list info api\n                    or (\n                        \"TimeLimit.truncated\" not in infos\n                        or (\n                            \"TimeLimit.truncated\" in infos\n                            and not infos[\"TimeLimit.truncated\"][i]\n                        )\n                    )\n                    # vector env, dict info api, for env i, vector mask `_TimeLimit.truncated` is not considered, to be compatible with envpool\n                    # For env i, `TimeLimit.truncated` not being present is treated same as being present and set to False.\n                    # therefore, terminated=True, truncated=True simultaneously is not allowed while using compatibility functions\n                    # with vector info\n                )\n            ):\n                terminateds.append(dones[i])\n                truncateds.append(False)\n\n            # This means info[\"TimeLimit.truncated\"] exists and this elif checks if it is True, which means the truncation has occurred but termination has not.\n            elif (\n                infos[\"TimeLimit.truncated\"]\n                if not is_vector_env\n                else (\n                    infos[\"TimeLimit.truncated\"][i]\n                    if isinstance(infos, dict)\n                    else infos[i][\"TimeLimit.truncated\"]\n                )\n            ):\n                assert dones[i]\n                terminateds.append(False)\n                truncateds.append(True)\n            else:\n                # This means info[\"TimeLimit.truncated\"] exists but is False, which means the core environment had already terminated,\n                # but it also exceeded maximum timesteps at the same step. However to be compatible with envpool, and to be backward compatible\n                # truncated is set to False here.\n                assert dones[i]\n                terminateds.append(True)\n                truncateds.append(False)\n\n        return (\n            observations,\n            rewards,\n            np.array(terminateds, dtype=np.bool_) if is_vector_env else terminateds[0],\n            np.array(truncateds, dtype=np.bool_) if is_vector_env else truncateds[0],\n            infos,\n        )\n\n\ndef step_to_old_api(\n    step_returns: Union[NewStepType, OldStepType], is_vector_env: bool = False\n) -> OldStepType:\n    \"\"\"Function to transform step returns to old step API irrespective of input API.\n\n    Args:\n        step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n        is_vector_env (bool): Whether the step_returns are from a vector environment\n    \"\"\"\n    if len(step_returns) == 4:\n        return step_returns\n    else:\n        assert len(step_returns) == 5\n        observations, rewards, terminateds, truncateds, infos = step_returns\n        dones = []\n        if not is_vector_env:\n            terminateds = [terminateds]\n            truncateds = [truncateds]\n\n        n_envs = len(terminateds)\n\n        for i in range(n_envs):\n            dones.append(terminateds[i] or truncateds[i])\n            if truncateds[i]:\n                if is_vector_env:\n                    # handle vector infos for dict and list\n                    if isinstance(infos, dict):\n                        if \"TimeLimit.truncated\" not in infos:\n                            # TODO: This should ideally not be done manually and should use vector_env's _add_info()\n                            infos[\"TimeLimit.truncated\"] = np.zeros(n_envs, dtype=bool)\n                            infos[\"_TimeLimit.truncated\"] = np.zeros(n_envs, dtype=bool)\n\n                        infos[\"TimeLimit.truncated\"][i] = (\n                            not terminateds[i] or infos[\"TimeLimit.truncated\"][i]\n                        )\n                        infos[\"_TimeLimit.truncated\"][i] = True\n                    else:\n                        # if vector info is a list\n                        infos[i][\"TimeLimit.truncated\"] = not terminateds[i] or infos[\n                            i\n                        ].get(\"TimeLimit.truncated\", False)\n                else:\n                    infos[\"TimeLimit.truncated\"] = not terminateds[i] or infos.get(\n                        \"TimeLimit.truncated\", False\n                    )\n        return (\n            observations,\n            rewards,\n            np.array(dones, dtype=np.bool_) if is_vector_env else dones[0],\n            infos,\n        )\n\n\ndef step_api_compatibility(\n    step_returns: Union[NewStepType, OldStepType],\n    new_step_api: bool = False,\n    is_vector_env: bool = False,\n) -> Union[NewStepType, OldStepType]:\n    \"\"\"Function to transform step returns to the API specified by `new_step_api` bool.\n\n    Old step API refers to step() method returning (observation, reward, done, info)\n    New step API refers to step() method returning (observation, reward, terminated, truncated, info)\n    (Refer to docs for details on the API change)\n\n    Args:\n        step_returns (tuple): Items returned by step(). Can be (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n        new_step_api (bool): Whether the output should be in new step API or old (False by default)\n        is_vector_env (bool): Whether the step_returns are from a vector environment\n\n    Returns:\n        step_returns (tuple): Depending on `new_step_api` bool, it can return (obs, rew, done, info) or (obs, rew, terminated, truncated, info)\n\n    Examples:\n        This function can be used to ensure compatibility in step interfaces with conflicting API. Eg. if env is written in old API,\n         wrapper is written in new API, and the final step output is desired to be in old API.\n\n        >>> obs, rew, done, info = step_api_compatibility(env.step(action))\n        >>> obs, rew, terminated, truncated, info = step_api_compatibility(env.step(action), new_step_api=True)\n        >>> observations, rewards, dones, infos = step_api_compatibility(vec_env.step(action), is_vector_env=True)\n    \"\"\"\n    if new_step_api:\n        return step_to_new_api(step_returns, is_vector_env)\n    else:\n        return step_to_old_api(step_returns, is_vector_env)\n", "path": "gym/utils/step_api_compatibility.py"}]}