class_docstr: Base class for memories organized as a queue / circular buffer.
  
  Args:
      capacity (int > 0): Memory capacity
          (<span style="color:#00C000"><b>default</b></span>: minimum capacity).
      device (string): Device name
          (<span style="color:#00C000"><b>default</b></span>: CPU:0).
      name (string): <span style="color:#0000C0"><b>internal use</b></span>.
      values_spec (specification): <span style="color:#0000C0"><b>internal use</b></span>.
      min_capacity (int >= 0): <span style="color:#0000C0"><b>internal use</b></span>.
class_docstr_tok: ["Base","class","for","memories","organized","as","a","queue","\/","circular","buffer",".","Args",":","capacity","(","int",">","0",")",":","Memory","capacity","(","<","span","style=","''","color",":","#","00C000","''",">","<","b",">","default","<","\/b",">","<","\/span",">",":","minimum","capacity",")",".","device","(","string",")",":","Device","name","(","<","span","style=","''","color",":","#","00C000","''",">","<","b",">","default","<","\/b",">","<","\/span",">",":","CPU:0",")",".","name","(","string",")",":","<","span","style=","''","color",":","#","0000C0","''",">","<","b",">","internal","use","<","\/b",">","<","\/span",">",".","values_spec","(","specification",")",":","<","span","style=","''","color",":","#","0000C0","''",">","<","b",">","internal","use","<","\/b",">","<","\/span",">",".","min_capacity","(","int",">","=","0",")",":","<","span","style=","''","color",":","#","0000C0","''",">","<","b",">","internal","use","<","\/b",">","<","\/span",">","."]
code: def enqueue(
      self, *, states, internals, auxiliaries, actions, terminal, reward
  ):
      zero = tf_util.constant(value=0, dtype="int")
      one = tf_util.constant(value=1, dtype="int")
      three = tf_util.constant(value=3, dtype="int")
      capacity = tf_util.constant(value=self.capacity, dtype="int")
      num_timesteps = tf_util.cast(
          x=tf.shape(input=terminal)[0], dtype="int"
      )
  
      last_index = tf.math.mod(x=(self.buffer_index - one), y=capacity)
  
      def correct_terminal():
          # Remove last observation terminal marker
          updates = tf_util.constant(value=0, dtype="int", shape=(1,))
          indices = tf.expand_dims(
              input=tf.expand_dims(input=last_index, axis=0), axis=1
          )
          value = tf.tensor_scatter_nd_update(
              tensor=self.buffers["terminal"],
              indices=indices,
              updates=updates,
          )
          assignment = self.buffers["terminal"].assign(value=value)
          # sparse_delta = tf.IndexedSlices(values=zero, indices=last_index)
          # assignment = self.buffers['terminal'].scatter_update(sparse_delta=sparse_delta)
          with tf.control_dependencies(control_inputs=(assignment,)):
              return last_index < zero
  
      last_terminal = tf.gather(
          params=self.buffers["terminal"], indices=last_index
      )
      is_incorrect = tf.math.equal(x=last_terminal, y=three)
      corrected = tf.cond(
          pred=is_incorrect, true_fn=correct_terminal, false_fn=tf.no_op
      )
  
      # Assertions
      assertions = [corrected]
      if self.config.create_tf_assertions:
          with tf.control_dependencies(control_inputs=(corrected,)):
              # check: number of timesteps fit into effectively available buffer
              assertions.append(
                  tf.debugging.assert_less_equal(
                      x=num_timesteps,
                      y=capacity,
                      message="Memory does not have enough capacity.",
                  )
              )
              # at most one terminal
              assertions.append(
                  tf.debugging.assert_less_equal(
                      x=tf.math.count_nonzero(
                          input=terminal,
                          dtype=tf_util.get_dtype(type="int"),
                      ),
                      y=one,
                      message="Timesteps contain more than one terminal.",
                  )
              )
              # if terminal, last timestep in batch
              assertions.append(
                  tf.debugging.assert_equal(
                      x=tf.math.reduce_any(
                          input_tensor=tf.math.greater(
                              x=terminal, y=zero
                          )
                      ),
                      y=tf.math.greater(x=terminal[-1], y=zero),
                      message="Terminal is not the last timestep.",
                  )
              )
              # general check: all terminal indices true
              assertions.append(
                  tf.debugging.assert_equal(
                      x=tf.reduce_all(
                          input_tensor=tf.gather(
                              params=tf.math.greater(
                                  x=self.buffers["terminal"], y=zero
                              ),
                              indices=self.terminal_indices[
                                  : self.episode_count + one
                              ],
                          )
                      ),
                      y=tf_util.constant(value=True, dtype="bool"),
                      message="Memory consistency check.",
                  )
              )
              # general check: only terminal indices true
              assertions.append(
                  tf.debugging.assert_equal(
                      x=tf.math.count_nonzero(
                          input=self.buffers["terminal"],
                          dtype=tf_util.get_dtype(type="int"),
                      ),
                      y=(self.episode_count + one),
                      message="Memory consistency check.",
                  )
              )
  
      # Buffer indices to overwrite
      with tf.control_dependencies(control_inputs=assertions):
          overwritten_indices = tf.range(
              start=self.buffer_index,
              limit=(self.buffer_index + num_timesteps),
          )
          overwritten_indices = tf.math.mod(
              x=overwritten_indices, y=capacity
          )
  
          # Count number of overwritten episodes
          num_episodes = tf.math.count_nonzero(
              input=tf.gather(
                  params=self.buffers["terminal"],
                  indices=overwritten_indices,
              ),
              axis=0,
              dtype=tf_util.get_dtype(type="int"),
          )
  
          # Shift remaining terminal indices accordingly
          index = self.episode_count + one
          assertions = list()
          if self.config.create_tf_assertions:
              assertions.append(
                  tf.debugging.assert_greater_equal(
                      x=index,
                      y=num_episodes,
                      message="Memory episode overwriting check.",
                  )
              )
  
      with tf.control_dependencies(control_inputs=assertions):
          updates = self.terminal_indices[num_episodes:index]
          indices = tf.expand_dims(
              input=tf.range(index - num_episodes), axis=1
          )
          value = tf.tensor_scatter_nd_update(
              tensor=self.terminal_indices,
              indices=indices,
              updates=updates,
          )
          assignment = self.terminal_indices.assign(value=value)
          # sparse_delta = tf.IndexedSlices(
          #     values=self.terminal_indices[num_episodes: index],
          #     indices=tf.range(index - num_episodes)
          # )
          # assignment = self.terminal_indices.scatter_update(sparse_delta=sparse_delta)
  
      # Decrement episode count accordingly
      with tf.control_dependencies(control_inputs=(assignment,)):
          assignment = self.episode_count.assign_sub(
              delta=num_episodes, read_value=False
          )
  
      # Write new observations
      with tf.control_dependencies(control_inputs=(assignment,)):
          # Add last observation terminal marker
          corrected_terminal = tf.where(
              condition=tf.math.equal(x=terminal[-1:], y=zero),
              x=three,
              y=terminal[-1:],
          )
          corrected_terminal = tf.concat(
              values=(terminal[:-1], corrected_terminal), axis=0
          )
          values = TensorDict(
              states=states,
              internals=internals,
              auxiliaries=auxiliaries,
              actions=actions,
              terminal=corrected_terminal,
              reward=reward,
          )
          indices = tf.range(
              start=self.buffer_index,
              limit=(self.buffer_index + num_timesteps),
          )
          indices = tf.math.mod(x=indices, y=capacity)
          indices = tf.expand_dims(input=indices, axis=1)
  
          def function(buffer, value):
              value = tf.tensor_scatter_nd_update(
                  tensor=buffer, indices=indices, updates=value
              )
              return buffer.assign(value=value)
              # sparse_delta = tf.IndexedSlices(values=value, indices=indices)
              # return buffer.scatter_update(sparse_delta=sparse_delta)
  
          assignments = self.buffers.fmap(
              function=function, cls=list, zip_values=values
          )
  
      # Increment buffer index
      with tf.control_dependencies(control_inputs=assignments):
          assignment = self.buffer_index.assign_add(
              delta=num_timesteps, read_value=False
          )
  
      # Count number of new episodes
      with tf.control_dependencies(control_inputs=(assignment,)):
          num_new_episodes = tf.math.count_nonzero(
              input=terminal, dtype=tf_util.get_dtype(type="int")
          )
  
          # Write new terminal indices
          new_terminal_indices = tf.boolean_mask(
              tensor=overwritten_indices,
              mask=tf.math.greater(x=terminal, y=zero),
          )
          start = self.episode_count + one
          indices = tf.expand_dims(
              input=tf.range(
                  start=start, limit=(start + num_new_episodes)
              ),
              axis=1,
          )
          value = tf.tensor_scatter_nd_update(
              tensor=self.terminal_indices,
              indices=indices,
              updates=new_terminal_indices,
          )
          assignment = self.terminal_indices.assign(value=value)
          # sparse_delta = tf.IndexedSlices(
          #     values=new_terminal_indices,
          #     indices=tf.range(start=start, limit=(start + num_new_episodes))
          # )
          # assignment = self.terminal_indices.scatter_update(sparse_delta=sparse_delta)
  
      # Increment episode count accordingly
      with tf.control_dependencies(control_inputs=(assignment,)):
          assignment = self.episode_count.assign_add(
              delta=num_new_episodes
          )
          return assignment < zero
  
file_path: tensorforce/tensorforce/core/memories/queue.py
imports: import numpy
  import tensorflow
  from tensorforce import TensorforceError, util
  from tensorforce.core import TensorDict, TensorSpec, tf_function, tf_util, VariableDict
  from tensorforce.core.memories import Memory
label_desc: Use image node_id 4 for calling the Queue obj's underlying member method code with example usage: obj.enqueue() and returns: unknown, unknown, buffer