docstr: A very minimal test that checks wheter pipeline connectors work in a
  simulation rollout.
docstr_tok: ["A","very","minimal","test","that","checks","wheter","pipeline","connectors","work","in","a","simulation","rollout","."]
code: def test_connector_pipline_with_view_requirement(self):
      """A very minimal test that checks wheter pipeline connectors work in a
      simulation rollout."""
      # TODO: make this test beefier and more comprehensive
      config = (
          PPOConfig()
          .framework("torch")
          .environment(env="CartPole-v1")
          .rollouts(create_env_on_local_worker=True)
      )
  
      env = gym.make("CartPole-v1")
      policy = config.build().get_policy()
  
      REQUIRED_KEYS = {
          SampleBatch.OBS,
          SampleBatch.NEXT_OBS,
          SampleBatch.REWARDS,
          SampleBatch.TERMINATEDS,
          SampleBatch.TRUNCATEDS,
          SampleBatch.INFOS,
          SampleBatch.ACTIONS,
      }
      policy.view_requirements = {
          k: v
          for k, v in policy.view_requirements.items()
          if k in REQUIRED_KEYS
      }
  
      # create a connector context
      ctx = ConnectorContext(
          view_requirements=policy.view_requirements,
          config=policy.config,
          initial_states=policy.get_initial_state(),
          is_policy_recurrent=policy.is_recurrent(),
          observation_space=policy.observation_space,
          action_space=policy.action_space,
      )
  
      # build chain of connectors
      connectors = [
          ObsPreprocessorConnector(ctx),
          StateBufferConnector(ctx),
          ViewRequirementAgentConnector(ctx),
      ]
      agent_connector = AgentConnectorPipeline(ctx, connectors)
  
      name, params = agent_connector.to_state()
      restored = get_connector(name, ctx, params)
      self.assertTrue(isinstance(restored, AgentConnectorPipeline))
      for cidx, c in enumerate(connectors):
          check(restored.connectors[cidx].to_state(), c.to_state())
  
      # simulate a rollout
      n_steps = 10
      obs, info = env.reset()
      env_out = AgentConnectorDataType(
          0, 1, {SampleBatch.NEXT_OBS: obs, SampleBatch.T: -1}
      )
      agent_obs = agent_connector([env_out])[0]
      t = 0
      total_rewards = 0
      while t < n_steps:
          policy_output = policy.compute_actions_from_input_dict(
              agent_obs.data.sample_batch
          )
          # Removes batch dimension
          policy_output = tree.map_structure(
              lambda x: x[0], policy_output
          )
  
          agent_connector.on_policy_output(
              ActionConnectorDataType(0, 1, {}, policy_output)
          )
          action = policy_output[0]
  
          next_obs, rewards, terminateds, truncateds, info = env.step(
              action
          )
          env_out_dict = {
              SampleBatch.NEXT_OBS: next_obs,
              SampleBatch.REWARDS: rewards,
              SampleBatch.TERMINATEDS: terminateds,
              SampleBatch.TRUNCATEDS: truncateds,
              SampleBatch.INFOS: info,
              SampleBatch.ACTIONS: action,
              # state_out
          }
          env_out = AgentConnectorDataType(0, 1, env_out_dict)
          agent_obs = agent_connector([env_out])[0]
          total_rewards += rewards
          t += 1
      print(total_rewards)
  
imports: import gymnasium
  from gymnasium.spaces import Box
  import numpy
  import tree
  import unittest
  from ray.rllib.algorithms.ppo.ppo import PPOConfig
  from ray.rllib.connectors.agent.clip_reward import ClipRewardAgentConnector
  from ray.rllib.connectors.agent.lambdas import FlattenDataAgentConnector
  from ray.rllib.connectors.agent.obs_preproc import ObsPreprocessorConnector
  from ray.rllib.connectors.agent.pipeline import AgentConnectorPipeline
  from ray.rllib.connectors.agent.state_buffer import StateBufferConnector
  from ray.rllib.connectors.agent.view_requirement import ViewRequirementAgentConnector
  from ray.rllib.connectors.connector import ConnectorContext
  from ray.rllib.connectors.registry import get_connector
  from ray.rllib.policy.view_requirement import ViewRequirement
  from ray.rllib.policy.sample_batch import SampleBatch
  from ray.rllib.utils.test_utils import check
  from ray.rllib.utils.typing import ActionConnectorDataType, AgentConnectorDataType, AgentConnectorsOutput
  from ray.rllib.connectors.agent.mean_std_filter import MeanStdObservationFilterAgentConnector
label_desc: Use image node_id 5 for calling the TestViewRequirementAgentConnector obj's underlying member method code with example usage: obj.test_connector_pipline_with_view_requirement() without return types