class_docstr: Stores multi-agent episode data.
  
  The central attribute of the class is the timestep mapping
  `global_t_to_local_t` that maps the global (environment)
  timestep to the local (agent) timesteps.
  
  The `MultiAgentEpisode` is based on the `SingleAgentEpisode`s
  for each agent, stored in `MultiAgentEpisode.agent_episodes`.
class_docstr_tok: ["Stores","multi-agent","episode","data",".","The","central","attribute","of","the","class","is","the","timestep","mapping","`","global_t_to_local_t","`","that","maps","the","global","(","environment",")","timestep","to","the","local","(","agent",")","timesteps",".","The","`","MultiAgentEpisode","`","is","based","on","the","`","SingleAgentEpisode","`","s","for","each","agent",",","stored","in","`","MultiAgentEpisode.agent_episodes","`","."]
docstr: Generates a SingleAgentEpisode from multi-agent data.
  
  Note, if no data is provided an empty `SingleAgentEpiosde`
  will be returned that starts at `SingleAgentEpisode.t_started=0`.
code: def _generate_single_agent_episode(
      self,
      agent_id: str,
      agent_episode_ids: Optional[Dict[str, str]] = None,
      observations: Optional[List[MultiAgentDict]] = None,
      actions: Optional[List[MultiAgentDict]] = None,
      rewards: Optional[List[MultiAgentDict]] = None,
      infos: Optional[List[MultiAgentDict]] = None,
      terminateds: Union[MultiAgentDict, bool] = False,
      truncateds: Union[MultiAgentDict, bool] = False,
      extra_model_outputs: Optional[List[MultiAgentDict]] = None,
  ) -> SingleAgentEpisode:
      """Generates a SingleAgentEpisode from multi-agent data.
  
      Note, if no data is provided an empty `SingleAgentEpiosde`
      will be returned that starts at `SingleAgentEpisode.t_started=0`.
      """
  
      # If an episode id for an agent episode was provided assign it.
      episode_id = (
          None
          if agent_episode_ids is None
          else agent_episode_ids[agent_id]
      )
      # We need the timestep mapping to create single agent's episode.
      if len(self.global_t_to_local_t) > 0:
          # Set to None if not provided.
          agent_observations = (
              None
              if observations is None
              else self._get_single_agent_data(
                  agent_id, observations, shift=-self.ts_carriage_return
              )
          )
  
          agent_actions = (
              None
              if actions is None
              else self._get_single_agent_data(
                  agent_id,
                  actions,
                  use_global_t_to_local_t=False,
              )
          )
  
          # Rewards are complicated in multi-agent scenarios, as agents could receive
          # a reward even though they did not get an observation or stepped at a
          # certain timestep.
          agent_rewards = (
              None
              if rewards is None
              else self._get_single_agent_data(
                  agent_id,
                  rewards,
                  use_global_t_to_local_t=False,
              )
          )
          # Like observations, infos start at timestep `t=0`, so we do not need to
          # shift or start later when using the global timestep mapping. But we
          # need to use the timestep carriage in case the starting timestep is
          # different from the length of observations-after-initialization.
          agent_infos = (
              None
              if infos is None
              else self._get_single_agent_data(
                  agent_id, infos, shift=-self.ts_carriage_return
              )
          )
  
          _agent_extra_model_outputs = (
              None
              if extra_model_outputs is None
              else self._get_single_agent_data(
                  agent_id,
                  extra_model_outputs,
                  use_global_t_to_local_t=False,
              )
          )
          # Convert `extra_model_outputs` for this agent from list of dicts to dict
          # of lists.
          agent_extra_model_outputs = defaultdict(list)
          if _agent_extra_model_outputs:
              for _model_out in _agent_extra_model_outputs:
                  for key, val in _model_out.items():
                      agent_extra_model_outputs[key].append(val)
  
          agent_is_terminated = terminateds.get(agent_id, False)
          agent_is_truncated = truncateds.get(agent_id, False)
  
          # If there are as many actions as observations we have to buffer.
          if (
              agent_actions
              and agent_observations
              and len(agent_observations) == len(agent_actions)
          ):
              # Assert then that the other data is in order.
              if agent_extra_model_outputs:
                  assert all(
                      len(v) == len(agent_actions)
                      for v in agent_extra_model_outputs.values()
                  ), (
                      f"Agent {agent_id} doesn't have the same number of "
                      "`extra_model_outputs` as it has actions "
                      f"({len(agent_actions)})."
                  )
                  # Put the last extra model outputs into the buffer.
                  self.agent_buffers[agent_id][
                      "extra_model_outputs"
                  ].get_nowait()
                  self.agent_buffers[agent_id][
                      "extra_model_outputs"
                  ].put_nowait(
                      {
                          k: v.pop()
                          for k, v in agent_extra_model_outputs.items()
                      }
                  )
  
              # Put the last action into the buffer.
              self.agent_buffers[agent_id]["actions"].put_nowait(
                  agent_actions.pop()
              )
  
          # TODO (simon): Check, if this can be refactored to a
          # `_generate_partial_rewards` method and can be done where
          # the global timestep  and global action timestep
          # mappings are created (__init__).
          # We have to take care of partial rewards when generating the agent rewards:
          #   1. Rewards between different observations -> added up and
          #       assigned to next observation.
          #   2. Rewards after the last observation -> get buffered and added up
          #       in the buffer for the next observation.
          #   3. Rewards before the initial observation -> get buffered
          #       and added to the next observation.
          # All partial rewards are recorded in `partial_rewards` together
          # with their corresponding timesteps in `partial_rewards_t`.
          if agent_rewards and observations:
              partial_agent_rewards_t = _IndexMapping()
              partial_agent_rewards = []
              agent_rewards = []
              agent_reward = 0.0
              for t, reward in enumerate(rewards):
                  if agent_id in reward:
                      # Add the rewards
                      partial_agent_rewards.append(reward[agent_id])
                      # Then add the reward.
                      agent_reward += reward[agent_id]
                      # Note, rewards start at timestep 1 (there are no initial ones).
                      # TODO (simon): Check, if we need to use here also
                      # `ts_carriage_return`.
                      partial_agent_rewards_t.append(
                          t + self.ts_carriage_return + 1
                      )
                      if (
                          t + self.ts_carriage_return + 1
                      ) in self.global_t_to_local_t[agent_id][1:]:
                          agent_rewards.append(agent_reward)
                          agent_reward = 0.0
  
              # If the agent reward is not zero, we must have rewards that came
              # after the last observation. Then we buffer this reward.
              self.agent_buffers[agent_id]["rewards"].put_nowait(
                  self.agent_buffers[agent_id]["rewards"].get_nowait()
                  + agent_reward
              )
              # Now save away the original rewards and the reward timesteps.
              self.partial_rewards_t[agent_id] = partial_agent_rewards_t
              self.partial_rewards[agent_id] = partial_agent_rewards
  
          return SingleAgentEpisode(
              id_=episode_id,
              observations=agent_observations,
              actions=agent_actions,
              rewards=agent_rewards,
              infos=agent_infos,
              terminated=agent_is_terminated,
              truncated=agent_is_truncated,
              extra_model_outputs=agent_extra_model_outputs,
          )
      # Otherwise return empty `SingleAgentEpisode`.
      else:
          return SingleAgentEpisode(id_=episode_id)
  
imports: from collections import defaultdict
  from queue import Queue
  from typing import Any, Dict, List, Optional, Set, Union
  import uuid
  import numpy
  from ray.rllib.env.single_agent_episode import SingleAgentEpisode
  from ray.rllib.policy.sample_batch import MultiAgentBatch
  from ray.rllib.utils.typing import MultiAgentDict
label_desc: Use image node_id 23 for calling the MultiAgentEpisode obj's underlying member method code with example usage: obj._generate_single_agent_episode(agent_id, agent_episode_ids, observations, actions, rewards, infos, terminateds, truncateds, extra_model_outputs) and returns: SingleAgentEpisode, SingleAgentEpisode