class_docstr: BaseEnv that executes its sub environments as @ray.remote actors.
  
  This provides dynamic batching of inference as observations are returned
  from the remote simulator actors. Both single and multi-agent child envs
  are supported, and envs can be stepped synchronously or asynchronously.
  
  NOTE: This class implicitly assumes that the remote envs are gym.Env's
  
  You shouldn't need to instantiate this class directly. It's automatically
  inserted when you use the `remote_worker_envs=True` option in your
  Algorithm's config.
class_docstr_tok: ["BaseEnv","that","executes","its","sub","environments","as","@","ray.remote","actors",".","This","provides","dynamic","batching","of","inference","as","observations","are","returned","from","the","remote","simulator","actors",".","Both","single","and","multi-agent","child","envs","are","supported",",","and","envs","can","be","stepped","synchronously","or","asynchronously",".","NOTE",":","This","class","implicitly","assumes","that","the","remote","envs","are","gym.Env's","You","should","n't","need","to","instantiate","this","class","directly",".","It","'s","automatically","inserted","when","you","use","the","`","remote_worker_envs=True","`","option","in","your","Algorithm","'s","config","."]
code: def poll(
      self,
  ) -> Tuple[
      MultiEnvDict,
      MultiEnvDict,
      MultiEnvDict,
      MultiEnvDict,
      MultiEnvDict,
      MultiEnvDict,
  ]:
      # each keyed by env_id in [0, num_remote_envs)
      obs, rewards, terminateds, truncateds, infos = {}, {}, {}, {}, {}
      ready = []
  
      # Wait for at least 1 env to be ready here.
      while not ready:
          ready, _ = ray.wait(
              list(self.pending),
              num_returns=len(self.pending),
              timeout=self.poll_timeout,
          )
  
      # Get and return observations for each of the ready envs
      env_ids = set()
      for obj_ref in ready:
          # Get the corresponding actor handle from our dict and remove the
          # object ref (we will call `ray.get()` on it and it will no longer
          # be "pending").
          actor = self.pending.pop(obj_ref)
          env_id = self.actors.index(actor)
          env_ids.add(env_id)
          # Get the ready object ref (this may be return value(s) of
          # `reset()` or `step()`).
          try:
              ret = ray.get(obj_ref)
          except Exception as e:
              # Something happened on the actor during stepping/resetting.
              # Restart sub-environment (create new actor; close old one).
              if self.restart_failed_sub_environments:
                  logger.exception(e.args[0])
                  self.try_restart(env_id)
                  # Always return multi-agent data.
                  # Set the observation to the exception, no rewards,
                  # terminated[__all__]=True (episode will be discarded anyways),
                  # no infos.
                  ret = (
                      e,
                      {},
                      {"__all__": True},
                      {"__all__": False},
                      {},
                  )
              # Do not try to restart. Just raise the error.
              else:
                  raise e
  
          # Our sub-envs are simple Actor-turned gym.Envs or MultiAgentEnvs.
          if self.make_env_creates_actors:
              rew, terminated, truncated, info = None, None, None, None
              if self.multiagent:
                  if isinstance(ret, tuple):
                      # Gym >= 0.26: `step()` result: Obs, reward, terminated,
                      # truncated, info.
                      if len(ret) == 5:
                          ob, rew, terminated, truncated, info = ret
                      # Gym >= 0.26: `reset()` result: Obs and infos.
                      elif len(ret) == 2:
                          ob = ret[0]
                          info = ret[1]
                      # Gym < 0.26? Something went wrong.
                      else:
                          raise AssertionError(
                              "Your gymnasium.Env seems to NOT return the correct "
                              "number of return values for `step()` (needs to return"
                              " 5 values: obs, reward, terminated, truncated and "
                              "info) or `reset()` (needs to return 2 values: obs and "
                              "info)!"
                          )
                  # Gym < 0.26: `reset()` result: Only obs.
                  else:
                      raise AssertionError(
                          "Your gymnasium.Env seems to only return a single value "
                          "upon `reset()`! Must return 2 (obs AND infos)."
                      )
              else:
                  if isinstance(ret, tuple):
                      # `step()` result: Obs, reward, terminated, truncated, info.
                      if len(ret) == 5:
                          ob = {_DUMMY_AGENT_ID: ret[0]}
                          rew = {_DUMMY_AGENT_ID: ret[1]}
                          terminated = {
                              _DUMMY_AGENT_ID: ret[2],
                              "__all__": ret[2],
                          }
                          truncated = {
                              _DUMMY_AGENT_ID: ret[3],
                              "__all__": ret[3],
                          }
                          info = {_DUMMY_AGENT_ID: ret[4]}
                      # `reset()` result: Obs and infos.
                      elif len(ret) == 2:
                          ob = {_DUMMY_AGENT_ID: ret[0]}
                          info = {_DUMMY_AGENT_ID: ret[1]}
                      # Gym < 0.26? Something went wrong.
                      else:
                          raise AssertionError(
                              "Your gymnasium.Env seems to NOT return the correct "
                              "number of return values for `step()` (needs to return"
                              " 5 values: obs, reward, terminated, truncated and "
                              "info) or `reset()` (needs to return 2 values: obs and "
                              "info)!"
                          )
                  # Gym < 0.26?
                  else:
                      raise AssertionError(
                          "Your gymnasium.Env seems to only return a single value "
                          "upon `reset()`! Must return 2 (obs and infos)."
                      )
  
              # If this is a `reset()` return value, we only have the initial
              # observations and infos: Set rewards, terminateds, and truncateds to
              # dummy values.
              if rew is None:
                  rew = {agent_id: 0 for agent_id in ob.keys()}
                  terminated = {"__all__": False}
                  truncated = {"__all__": False}
  
          # Our sub-envs are auto-wrapped (by `_RemoteSingleAgentEnv` or
          # `_RemoteMultiAgentEnv`) and already behave like multi-agent
          # envs.
          else:
              ob, rew, terminated, truncated, info = ret
          obs[env_id] = ob
          rewards[env_id] = rew
          terminateds[env_id] = terminated
          truncateds[env_id] = truncated
          infos[env_id] = info
  
      logger.debug(f"Got obs batch for actors {env_ids}")
      return obs, rewards, terminateds, truncateds, infos, {}
  
imports: import gymnasium
  import logging
  from typing import Callable, Dict, List, Optional, Set, Tuple, TYPE_CHECKING
  import ray
  from ray.util import log_once
  from ray.rllib.env.base_env import BaseEnv, _DUMMY_AGENT_ID, ASYNC_RESET_RETURN
  from ray.rllib.utils.annotations import override, PublicAPI
  from ray.rllib.utils.typing import AgentID, EnvID, EnvType, MultiEnvDict
label_desc: Use image node_id 2 for calling the RemoteBaseEnv obj's underlying member method code with example usage: obj.poll() and returns: obs,  rewards,  terminateds,  truncateds,  infos,  dict