class_docstr: A CartPole env that crashes (or stalls) from time to time.
  
  Useful for testing faulty sub-env (within a vectorized env) handling by
  EnvRunners.
  
  After crashing, the env expects a `reset()` call next (calling `step()` will
  result in yet another error), which may or may not take a very long time to
  complete. This simulates the env having to reinitialize some sub-processes, e.g.
  an external connection.
  
  The env can also be configured to stall (and do nothing during a call to `step()`)
  from time to time for a configurable amount of time.
class_docstr_tok: ["A","CartPole","env","that","crashes","(","or","stalls",")","from","time","to","time",".","Useful","for","testing","faulty","sub-env","(","within","a","vectorized","env",")","handling","by","EnvRunners",".","After","crashing",",","the","env","expects","a","`","reset","(",")","`","call","next","(","calling","`","step","(",")","`","will","result","in","yet","another","error",")",",","which","may","or","may","not","take","a","very","long","time","to","complete",".","This","simulates","the","env","having","to","reinitialize","some","sub-processes",",","e.g",".","an","external","connection",".","The","env","can","also","be","configured","to","stall","(","and","do","nothing","during","a","call","to","`","step","(",")","`",")","from","time","to","time","for","a","configurable","amount","of","time","."]
code: def __init__(self, config=None):
      super().__init__()
  
      self.config = config if config is not None else {}
  
      # Crash probability (in each `step()`).
      self.p_crash = config.get("p_crash", 0.005)
      # Crash probability when `reset()` is called.
      self.p_crash_reset = config.get("p_crash_reset", 0.0)
      # Crash exactly after every n steps. If a 2-tuple, will uniformly sample
      # crash timesteps from in between the two given values.
      self.crash_after_n_steps = config.get("crash_after_n_steps")
      self._crash_after_n_steps = None
      assert (
          self.crash_after_n_steps is None
          or isinstance(self.crash_after_n_steps, int)
          or (
              isinstance(self.crash_after_n_steps, tuple)
              and len(self.crash_after_n_steps) == 2
          )
      )
      # Only ever crash, if on certain worker indices.
      faulty_indices = config.get("crash_on_worker_indices", None)
      if faulty_indices and config.worker_index not in faulty_indices:
          self.p_crash = 0.0
          self.p_crash_reset = 0.0
          self.crash_after_n_steps = None
  
      # Stall probability (in each `step()`).
      self.p_stall = config.get("p_stall", 0.0)
      # Stall probability when `reset()` is called.
      self.p_stall_reset = config.get("p_stall_reset", 0.0)
      # Stall exactly after every n steps.
      self.stall_after_n_steps = config.get("stall_after_n_steps")
      self._stall_after_n_steps = None
      # Amount of time to stall. If a 2-tuple, will uniformly sample from in between
      # the two given values.
      self.stall_time_sec = config.get("stall_time_sec")
      assert (
          self.stall_time_sec is None
          or isinstance(self.stall_time_sec, (int, float))
          or (
              isinstance(self.stall_time_sec, tuple)
              and len(self.stall_time_sec) == 2
          )
      )
  
      # Only ever stall, if on certain worker indices.
      faulty_indices = config.get("stall_on_worker_indices", None)
      if faulty_indices and config.worker_index not in faulty_indices:
          self.p_stall = 0.0
          self.p_stall_reset = 0.0
          self.stall_after_n_steps = None
  
      # Timestep counter for the ongoing episode.
      self.timesteps = 0
  
      # Time in seconds to initialize (in this c'tor).
      sample = 0.0
      if "init_time_s" in config:
          sample = (
              config["init_time_s"]
              if not isinstance(config["init_time_s"], tuple)
              else np.random.uniform(
                  config["init_time_s"][0], config["init_time_s"][1]
              )
          )
  
      print(
          f"Initializing crashing env (with init-delay of {sample}sec) ..."
      )
      time.sleep(sample)
  
      # No env pre-checking?
      self._skip_env_checking = config.get("skip_env_checking", False)
  
      # Make sure envs don't crash at the same time.
      self._rng = np.random.RandomState()
  
file_path: ray/rllib/examples/env/cartpole_crashing.py
imports: import logging
  from gymnasium.envs.classic_control import CartPoleEnv
  import numpy
  import time
  from ray.rllib.examples.env.multi_agent import make_multi_agent
  from ray.rllib.utils.annotations import override
  from ray.rllib.utils.error import EnvError
label_desc: Use image node_id 1 to create a new CartPoleCrashing object from inherited base classes: CartPoleEnv with example: obj = CartPoleCrashing(config)