Method read

dm_control/composer/observation/obs_buffer.py:162–171

Reads the content of the buffer at the given timestamp.

(self, current_time)

                       'got {!r}'.format(delay))

  def read(self, current_time):
    """Reads the content of the buffer at the given timestamp."""
    self._update_arrived_deque(current_time)
    if self._has_buffer_dim:
      out = np.empty(self._buffered_shape, dtype=self._dtype)
      for i, obs in enumerate(self._arrived_deque):
        out[i] = obs.value
    else:
      out = self._arrived_deque[0].value.copy()
    return out

  def drop_unobserved_upcoming_items(self, observation_schedule, read_interval):
    """Plans an optimal observation schedule for an upcoming control period.