class_docstr: Collect experiences from user environment using Connectors.
docstr: Process the output of policy neural network evaluation.
  
  Records policy evaluation results into agent connectors and
  returns replies to send back to agents in the env.
  
  Args:
      active_envs: Set of env IDs that are still active.
      to_eval: Mapping of policy IDs to lists of AgentConnectorDataType objects.
      eval_results: Mapping of policy IDs to list of
          actions, rnn-out states, extra-action-fetches dicts.
      off_policy_actions: Doubly keyed dict of env-ids -> agent ids ->
          off-policy-action, returned by a `BaseEnv.poll()` call.
  
  Returns:
      Nested dict of env id -> agent id -> actions to be sent to
      Env (np.ndarrays).
docstr_tok: ["Process","the","output","of","policy","neural","network","evaluation",".","Records","policy","evaluation","results","into","agent","connectors","and","returns","replies","to","send","back","to","agents","in","the","env",".","Args",":","active_envs",":","Set","of","env","IDs","that","are","still","active",".","to_eval",":","Mapping","of","policy","IDs","to","lists","of","AgentConnectorDataType","objects",".","eval_results",":","Mapping","of","policy","IDs","to","list","of","actions",",","rnn-out","states",",","extra-action-fetches","dicts",".","off_policy_actions",":","Doubly","keyed","dict","of","env-ids","-",">","agent","ids","-",">","off-policy-action",",","returned","by","a","`","BaseEnv.poll","(",")","`","call",".","Returns",":","Nested","dict","of","env","id","-",">","agent","id","-",">","actions","to","be","sent","to","Env","(","np.ndarrays",")","."]
code: def _process_policy_eval_results(
      self,
      active_envs: Set[EnvID],
      to_eval: Dict[PolicyID, List[AgentConnectorDataType]],
      eval_results: Dict[PolicyID, PolicyOutputType],
      off_policy_actions: MultiEnvDict,
  ):
      """Process the output of policy neural network evaluation.
  
      Records policy evaluation results into agent connectors and
      returns replies to send back to agents in the env.
  
      Args:
          active_envs: Set of env IDs that are still active.
          to_eval: Mapping of policy IDs to lists of AgentConnectorDataType objects.
          eval_results: Mapping of policy IDs to list of
              actions, rnn-out states, extra-action-fetches dicts.
          off_policy_actions: Doubly keyed dict of env-ids -> agent ids ->
              off-policy-action, returned by a `BaseEnv.poll()` call.
  
      Returns:
          Nested dict of env id -> agent id -> actions to be sent to
          Env (np.ndarrays).
      """
      actions_to_send: Dict[
          EnvID, Dict[AgentID, EnvActionType]
      ] = defaultdict(dict)
  
      for env_id in active_envs:
          actions_to_send[env_id] = {}  # at minimum send empty dict
  
      # types: PolicyID, List[AgentConnectorDataType]
      for policy_id, eval_data in to_eval.items():
          actions: TensorStructType = eval_results[policy_id][0]
          actions = convert_to_numpy(actions)
  
          rnn_out: StateBatches = eval_results[policy_id][1]
          extra_action_out: dict = eval_results[policy_id][2]
  
          # In case actions is a list (representing the 0th dim of a batch of
          # primitive actions), try converting it first.
          if isinstance(actions, list):
              actions = np.array(actions)
          # Split action-component batches into single action rows.
          actions: List[EnvActionType] = unbatch(actions)
  
          policy: Policy = _get_or_raise(
              self._worker.policy_map, policy_id
          )
          assert (
              policy.agent_connectors and policy.action_connectors
          ), "EnvRunnerV2 requires action connectors to work."
  
          # types: int, EnvActionType
          for i, action in enumerate(actions):
              env_id: int = eval_data[i].env_id
              agent_id: AgentID = eval_data[i].agent_id
              input_dict: TensorStructType = eval_data[i].data.raw_dict
  
              rnn_states: List[StateBatches] = tree.map_structure(
                  lambda x: x[i], rnn_out
              )
  
              # extra_action_out could be a nested dict
              fetches: Dict = tree.map_structure(
                  lambda x: x[i], extra_action_out
              )
  
              # Post-process policy output by running them through action connectors.
              ac_data = ActionConnectorDataType(
                  env_id,
                  agent_id,
                  input_dict,
                  (action, rnn_states, fetches),
              )
  
              (
                  action_to_send,
                  rnn_states,
                  fetches,
              ) = policy.action_connectors(ac_data).output
  
              # The action we want to buffer is the direct output of
              # compute_actions_from_input_dict() here. This is because we want to
              # send the unsqushed actions to the environment while learning and
              # possibly basing subsequent actions on the squashed actions.
              action_to_buffer = (
                  action
                  if env_id not in off_policy_actions
                  or agent_id not in off_policy_actions[env_id]
                  else off_policy_actions[env_id][agent_id]
              )
  
              # Notify agent connectors with this new policy output.
              # Necessary for state buffering agent connectors, for example.
              ac_data: ActionConnectorDataType = (
                  ActionConnectorDataType(
                      env_id,
                      agent_id,
                      input_dict,
                      (action_to_buffer, rnn_states, fetches),
                  )
              )
              policy.agent_connectors.on_policy_output(ac_data)
  
              assert agent_id not in actions_to_send[env_id]
              actions_to_send[env_id][agent_id] = action_to_send
  
      return actions_to_send
  
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 16 for calling the EnvRunnerV2 obj's underlying member method code with example usage: obj._process_policy_eval_results(active_envs, to_eval, eval_results, off_policy_actions) and returns: actions_to_send