Class SynchronizingArrayWrapper

dm_control/mjcf/physics.py:155–239

A non-contiguous view of an ndarray that synchronizes with the original. Note: this class should not be instantiated directly.

class SynchronizingArrayWrapper(np.ndarray):
  """A non-contiguous view of an ndarray that synchronizes with the original.

  Note: this class should not be instantiated directly.
  """
  __slots__ = (
      '_backing_array',
      '_backing_index',
      '_backing_index_is_array_like',
      '_physics',
      '_triggers_dirty',
      '_disable_on_write',
  )

  def __new__(cls,
              backing_array,
              backing_index,
              physics,
              triggers_dirty,
              disable_on_write):
    obj = backing_array[backing_index].view(SynchronizingArrayWrapper)
    # pylint: disable=protected-access
    obj._backing_array = backing_array
    obj._backing_index = backing_index
    # Performance optimization: avoid repeatedly checking the type of the
    # backing index.
    backing_index_type = _get_index_type(backing_index)
    obj._backing_index_is_array_like = backing_index_type is _ARRAY_LIKE
    obj._physics = physics
    obj._triggers_dirty = triggers_dirty
    obj._disable_on_write = disable_on_write
    # pylint: enable=protected-access
    return obj

  def _synchronize_from_backing_array(self):
    if self._physics.is_dirty and not self._triggers_dirty:
      self._physics.forward()
    updated_values = self._backing_array[self._backing_index]
    # Faster than `super(...).__setitem__(slice(None), updated_values)`
    np.copyto(self, updated_values)

  def copy(self, order='C'):
    return np.copy(self, order=order)

  def __copy__(self):
    return self.copy()

  def __deepcopy__(self, memo):
    return self.copy()

  def __reduce__(self):
    raise NotImplementedError(_PICKLING_NOT_SUPPORTED.format(type=type(self)))

  def __setitem__(self, index, value):
    if self._physics.is_dirty and not self._triggers_dirty:
      self._physics.forward()
    super().__setitem__(index, value)