class_docstr: Implement experience replay in the paper
  `Human-level control through deep reinforcement learning
  <http://www.nature.com/nature/journal/v518/n7540/full/nature14236.html>`_.
  
  This implementation provides the interface as a :class:`DataFlow`.
  This DataFlow is __not__ fork-safe (thus doesn't support multiprocess prefetching).
  
  It does the following:
  * Spawn `num_parallel_players` environment thread, each running an instance
    of the environment with epislon-greedy policy.
  * All environment instances writes their experiences to a shared replay
    memory buffer.
  * Produces batched samples by sampling the replay buffer. After producing
    each batch, it executes the environment instances by a total of
    `update_frequency` steps.
  
  This implementation assumes that state is batch-able, and the network takes batched inputs.
class_docstr_tok: ["Implement","experience","replay","in","the","paper","`","Human-level","control","through","deep","reinforcement","learning","<","http",":","\/\/www.nature.com\/nature\/journal\/v518\/n7540\/full\/nature14236.html",">","`","_",".","This","implementation","provides","the","interface","as","a",":","class",":","`","DataFlow","`",".","This","DataFlow","is","__not__","fork-safe","(","thus","does","n't","support","multiprocess","prefetching",")",".","It","does","the","following",":","*","Spawn","`","num_parallel_players","`","environment","thread",",","each","running","an","instance","of","the","environment","with","epislon-greedy","policy",".","*","All","environment","instances","writes","their","experiences","to","a","shared","replay","memory","buffer",".","*","Produces","batched","samples","by","sampling","the","replay","buffer",".","After","producing","each","batch",",","it","executes","the","environment","instances","by","a","total","of","`","update_frequency","`","steps",".","This","implementation","assumes","that","state","is","batch-able",",","and","the","network","takes","batched","inputs","."]
docstr: Args:
      predictor_io_names (tuple of list of str): input/output names to
          predict Q value from state.
      get_player (-> gym.Env): a callable which returns a player.
      num_parallel_players (int): number of players to run in parallel.
          Standard DQN uses 1.
          Parallelism increases speed, but will affect the distribution of
          experiences in the replay buffer.
      state_shape (tuple):
      batch_size (int):
      memory_size (int):
      init_memory_size (int):
      update_frequency (int): number of new transitions to add to memory
          after sampling a batch of transitions for training.
      history_len (int): length of history frames to concat. Zero-filled
          initial frames.
      state_dtype (str):
code: def __init__(
      self,
      predictor_io_names,
      get_player,
      num_parallel_players,
      state_shape,
      batch_size,
      memory_size,
      init_memory_size,
      update_frequency,
      history_len,
      state_dtype="uint8",
  ):
      """
      Args:
          predictor_io_names (tuple of list of str): input/output names to
              predict Q value from state.
          get_player (-> gym.Env): a callable which returns a player.
          num_parallel_players (int): number of players to run in parallel.
              Standard DQN uses 1.
              Parallelism increases speed, but will affect the distribution of
              experiences in the replay buffer.
          state_shape (tuple):
          batch_size (int):
          memory_size (int):
          init_memory_size (int):
          update_frequency (int): number of new transitions to add to memory
              after sampling a batch of transitions for training.
          history_len (int): length of history frames to concat. Zero-filled
              initial frames.
          state_dtype (str):
      """
      assert len(state_shape) in [1, 2, 3], state_shape
      init_memory_size = int(init_memory_size)
  
      for k, v in locals().items():
          if k != "self":
              setattr(self, k, v)
      self.exploration = 1.0  # default initial exploration
  
      self.rng = get_rng(self)
      self._init_memory_flag = (
          threading.Event()
      )  # tell if memory has been initialized
  
      self.mem = ReplayMemory(
          memory_size, state_shape, self.history_len, dtype=state_dtype
      )
  
imports: import copy
  import itertools
  import numpy
  import threading
  from collections import namedtuple
  from six.moves import queue, range
  from tensorpack.utils.concurrency import LoopThread, ShareSessionThread
  from tensorpack.callbacks.base import Callback
  from tensorpack.dataflow import DataFlow
  from tensorpack.utils import logger, get_rng, get_tqdm
  from tensorpack.utils.stats import StatCounter
label_desc: Use image node_id 1 to create a new ExpReplay object from inherited base classes: DataFlow, Callback with example: obj = ExpReplay(predictor_io_names, get_player, num_parallel_players, state_shape, batch_size, memory_size, init_memory_size, update_frequency, history_len, state_dtype)