class_docstr: Collect experiences from user environment using Connectors.
docstr: Call compute_actions on collected episode data to get next action.
  
  Args:
      to_eval: Mapping of policy IDs to lists of AgentConnectorDataType objects
          (items in these lists will be the batch's items for the model
          forward pass).
  
  Returns:
      Dict mapping PolicyIDs to compute_actions_from_input_dict() outputs.
docstr_tok: ["Call","compute_actions","on","collected","episode","data","to","get","next","action",".","Args",":","to_eval",":","Mapping","of","policy","IDs","to","lists","of","AgentConnectorDataType","objects","(","items","in","these","lists","will","be","the","batch","'s","items","for","the","model","forward","pass",")",".","Returns",":","Dict","mapping","PolicyIDs","to","compute_actions_from_input_dict","(",")","outputs","."]
code: def _do_policy_eval(
      self,
      to_eval: Dict[PolicyID, List[AgentConnectorDataType]],
  ) -> Dict[PolicyID, PolicyOutputType]:
      """Call compute_actions on collected episode data to get next action.
  
      Args:
          to_eval: Mapping of policy IDs to lists of AgentConnectorDataType objects
              (items in these lists will be the batch's items for the model
              forward pass).
  
      Returns:
          Dict mapping PolicyIDs to compute_actions_from_input_dict() outputs.
      """
      policies = self._worker.policy_map
  
      # In case policy map has changed, try to find the new policy that
      # should handle all these per-agent eval data.
      # Throws exception if these agents are mapped to multiple different
      # policies now.
      def _try_find_policy_again(eval_data: AgentConnectorDataType):
          policy_id = None
          for d in eval_data:
              episode = self._active_episodes[d.env_id]
              # Force refresh policy mapping on the episode.
              pid = episode.policy_for(d.agent_id, refresh=True)
              if policy_id is not None and pid != policy_id:
                  raise ValueError(
                      "Policy map changed. The list of eval data that was handled "
                      f"by a same policy is now handled by policy {pid} "
                      "and {policy_id}. "
                      "Please don't do this in the middle of an episode."
                  )
              policy_id = pid
          return _get_or_raise(self._worker.policy_map, policy_id)
  
      eval_results: Dict[PolicyID, TensorStructType] = {}
      for policy_id, eval_data in to_eval.items():
          # In case the policyID has been removed from this worker, we need to
          # re-assign policy_id and re-lookup the Policy object to use.
          try:
              policy: Policy = _get_or_raise(policies, policy_id)
          except ValueError:
              # policy_mapping_fn from the worker may have already been
              # changed (mapping fn not staying constant within one episode).
              policy: Policy = _try_find_policy_again(eval_data)
  
          if policy.config.get("_enable_new_api_stack", False):
              # _batch_inference_sample_batches does nothing but concatenating AND
              # setting SEQ_LENS to ones in the recurrent case. We do not need this
              # because RLModules do not care about SEQ_LENS anymore. They have an
              # expected input shape convention of [B, T, ...]
              input_dict = concat_samples(
                  [d.data.sample_batch for d in eval_data]
              )
          else:
              input_dict = _batch_inference_sample_batches(
                  [d.data.sample_batch for d in eval_data]
              )
  
          eval_results[
              policy_id
          ] = policy.compute_actions_from_input_dict(
              input_dict,
              timestep=policy.global_timestep,
              episodes=[
                  self._active_episodes[t.env_id] for t in eval_data
              ],
          )
  
      return eval_results
  
imports: from collections import defaultdict
  import logging
  import time
  import tree
  from typing import TYPE_CHECKING, Dict, Iterator, List, Optional, Set, Tuple, Union
  import numpy
  from ray.rllib.env.base_env import ASYNC_RESET_RETURN, BaseEnv
  from ray.rllib.env.external_env import ExternalEnvWrapper
  from ray.rllib.env.wrappers.atari_wrappers import MonitorEnv, get_wrapper_by_cls
  from ray.rllib.evaluation.collectors.simple_list_collector import _PolicyCollectorGroup
  from ray.rllib.policy.rnn_sequencing import pad_batch_to_sequences_of_same_size
  from ray.rllib.evaluation.episode_v2 import EpisodeV2
  from ray.rllib.evaluation.metrics import RolloutMetrics
  from ray.rllib.models.preprocessors import Preprocessor
  from ray.rllib.policy.policy import Policy
  from ray.rllib.policy.sample_batch import MultiAgentBatch, SampleBatch, concat_samples
  from ray.rllib.utils.annotations import DeveloperAPI
  from ray.rllib.utils.filter import Filter
  from ray.rllib.utils.numpy import convert_to_numpy
  from ray.rllib.utils.spaces.space_utils import unbatch, get_original_space
  from ray.rllib.utils.typing import ActionConnectorDataType, AgentConnectorDataType, AgentID, EnvActionType, EnvID, EnvInfoDict, EnvObsType, MultiAgentDict, MultiEnvDict, PolicyID, PolicyOutputType, SampleBatchType, StateBatches, TensorStructType
  from ray.util.debug import log_once
label_desc: Use image node_id 15 for calling the EnvRunnerV2 obj's underlying member method code with example usage: obj._do_policy_eval(to_eval) and returns: eval_results, _get_or_raise