class_docstr: A connector piece that stacks the previous n observations into one.
class_docstr_tok: ["A","connector","piece","that","stacks","the","previous","n","observations","into","one","."]
code: def __call__(
      self,
      *,
      rl_module: RLModule,
      data: Optional[Any],
      episodes: List[EpisodeType],
      explore: Optional[bool] = None,
      shared_data: Optional[dict] = None,
      **kwargs,
  ) -> Any:
      # This is a data-in-data-out connector, so we expect `data` to be a dict
      # with: key=column name, e.g. "obs" and value=[data to be processed by
      # RLModule]. We will add to `data` the last n observations.
      observations = []
  
      # Learner connector pipeline. Episodes have been finalized/numpy'ized.
      if self.as_learner_connector:
          for episode in episodes:
  
              def _map_fn(s):
                  # Squeeze out last dim.
                  s = np.squeeze(s, axis=-1)
                  # Calculate new shape and strides
                  new_shape = (len(episode), self.num_frames) + s.shape[
                      1:
                  ]
                  new_strides = (s.strides[0],) + s.strides
                  # Create a strided view of the array.
                  return np.lib.stride_tricks.as_strided(
                      s, shape=new_shape, strides=new_strides
                  )
  
              # Get all observations from the episode in one np array (except for
              # the very last one, which is the final observation not needed for
              # learning).
              observations.append(
                  tree.map_structure(
                      _map_fn,
                      episode.get_observations(
                          indices=slice(
                              -self.num_frames + 1, len(episode)
                          ),
                          neg_indices_left_of_zero=True,
                          fill=0.0,
                      ),
                  )
              )
  
          # Move stack-dimension to the end and concatenate along batch axis.
          data[SampleBatch.OBS] = tree.map_structure(
              lambda *s: np.transpose(
                  np.concatenate(s, axis=0), axes=[0, 2, 3, 1]
              ),
              *observations,
          )
  
      # Env-to-module pipeline. Episodes still operate on lists.
      else:
          for episode in episodes:
              assert not episode.is_finalized
              # Get the list of observations to stack.
              obs_stack = episode.get_observations(
                  indices=slice(-self.num_frames, None),
                  fill=0.0,
              )
              # Observation components are (w, h, 1)
              # -> stack to (w, h, [num_frames], 1), then squeeze out last dim to get
              # (w, h, [num_frames]).
              stacked_obs = tree.map_structure(
                  lambda *s: np.squeeze(np.stack(s, axis=2), axis=-1),
                  *obs_stack,
              )
              observations.append(stacked_obs)
  
          data[SampleBatch.OBS] = batch(observations)
  
      return data
  
file_path: ray/rllib/connectors/common/frame_stacking.py
imports: import numpy
  from typing import Any, List, Optional
  import gymnasium
  import tree
  from ray.rllib.connectors.connector_v2 import ConnectorV2
  from ray.rllib.core.rl_module.rl_module import RLModule
  from ray.rllib.policy.sample_batch import SampleBatch
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.spaces.space_utils import batch
  from ray.rllib.utils.typing import EpisodeType
label_desc: Use image node_id 2 for calling the _FrameStackingConnector obj's underlying member method code with example usage: obj.__call__() and returns: data, np