code: def get_cliff_walking_wall_policy_and_data(
      num_episodes: int, gamma: float, epsilon: float, seed: int
  ) -> Tuple[Policy, SampleBatch, float, float]:
      """Collect a cliff_walking_wall policy and data with epsilon-greedy exploration.
  
      Args:
          num_episodes: Minimum number of episodes to collect
          gamma: discount factor
          epsilon: epsilon-greedy exploration value
  
      Returns:
          A Tuple consisting of:
            - A CliffWalkingWallPolicy with exploration parameter epsilon
            - A SampleBatch of at least `num_episodes` CliffWalkingWall episodes
            collected using epsilon-greedy exploration
            - The mean of the discounted return over the collected episodes
            - The stddev of the discounted return over the collected episodes
  
      """
  
      config = (
          AlgorithmConfig()
          .debugging(seed=seed)
          .rollouts(batch_mode="complete_episodes")
          .environment(disable_env_checking=True)
          .experimental(_disable_preprocessor_api=True)
      )
      config = config.to_dict()
      config["epsilon"] = epsilon
  
      env = CliffWalkingWallEnv(seed=seed)
      policy = CliffWalkingWallPolicy(
          env.observation_space,
          env.action_space,
          {"epsilon": epsilon, "seed": seed},
      )
      workers = WorkerSet(
          env_creator=lambda env_config: CliffWalkingWallEnv(),
          default_policy_class=CliffWalkingWallPolicy,
          config=config,
          num_workers=4,
      )
      ep_ret = []
      batches = []
      n_eps = 0
      while n_eps < num_episodes:
          batch = synchronous_parallel_sample(worker_set=workers)
          batch = convert_ma_batch_to_sample_batch(batch)
  
          for episode in batch.split_by_episode():
              ret = 0
              for r in episode[SampleBatch.REWARDS][::-1]:
                  ret = r + gamma * ret
              ep_ret.append(ret)
              n_eps += 1
          batches.append(batch)
      workers.stop()
  
      return (
          policy,
          concat_samples(batches),
          np.mean(ep_ret),
          np.std(ep_ret),
      )
  
file_path: ray/python/ray/rllib/offline/estimators/tests/utils.py
imports: from typing import Type, Union, Dict, Tuple
  import numpy
  from ray.rllib.algorithms import AlgorithmConfig
  from ray.rllib.evaluation.worker_set import WorkerSet
  from ray.rllib.examples.env.cliff_walking_wall_env import CliffWalkingWallEnv
  from ray.rllib.examples.policy.cliff_walking_wall_policy import CliffWalkingWallPolicy
  from ray.rllib.execution.rollout_ops import synchronous_parallel_sample
  from ray.rllib.offline.estimators import DirectMethod, DoublyRobust
  from ray.rllib.policy import Policy
  from ray.rllib.policy.sample_batch import SampleBatch, concat_samples, convert_ma_batch_to_sample_batch
  from ray.rllib.utils.debug import update_global_seed_if_necessary
label_desc: Use image node_id 1 for calling a global function with example usage: get_cliff_walking_wall_policy_and_data(num_episodes, gamma, epsilon, seed) and returns: policy,  concat_samples,  np,  np