(self)
Simulates the next control step and optimizes the update schedule.
Simulates the next control step and optimizes the update schedule.
Simulates the next control step and optimizes the update schedule .
def prepare_for_next_control_step(self):
    """Simulates the next control step and optimizes the update schedule."""
    if self._enabled_structure is None:
      raise RuntimeError('`reset` must be called before `before_step`.')
    for enabled in self._enabled_list:

      if (enabled.update_interval == DEFAULT_UPDATE_INTERVAL
          and enabled.delay == DEFAULT_DELAY
          and enabled.buffer_size < self._physics_steps_per_control_step):
        for i in reversed(range(enabled.buffer_size)):
          next_step = (
              self._step_counter + self._physics_steps_per_control_step - i)
          next_delay = DEFAULT_DELAY
          enabled.update_schedule.append((next_step, next_delay))
      else:
        if enabled.update_schedule:
          last_scheduled_step = enabled.update_schedule[-1][0]
        else:
          last_scheduled_step = self._step_counter
        max_step = self._step_counter + 2 * self._physics_steps_per_control_step
        while last_scheduled_step < max_step:
          next_update_interval = _call_if_callable(enabled.update_interval)
          next_step = last_scheduled_step + next_update_interval
          next_delay = _call_if_callable(enabled.delay)
          enabled.update_schedule.append((next_step, next_delay))
          last_scheduled_step = next_step
        # Optimize the schedule by planning ahead and dropping unseen entries.
        enabled.buffer.drop_unobserved_upcoming_items(
            enabled.update_schedule, self._physics_steps_per_control_step)
def prepare_for_next_control_step ( self ) : if self . _enabled_structure is None : raise RuntimeError ( '`reset` must be called before `before_step`.' ) for enabled in self . _enabled_list : if ( enabled . update_interval == DEFAULT_UPDATE_INTERVAL and enabled . delay == DEFAULT_DELAY and enabled . buffer_size < self . _physics_steps_per_control_step ) : for i in reversed ( range ( enabled . buffer_size ) ) : next_step = ( self . _step_counter + self . _physics_steps_per_control_step - i ) next_delay = DEFAULT_DELAY enabled . update_schedule . append ( ( next_step , next_delay ) ) else : if enabled . update_schedule : last_scheduled_step = enabled . update_schedule [ - 1 ] [ 0 ] else : last_scheduled_step = self . _step_counter max_step = self . _step_counter + 2 * self . _physics_steps_per_control_step while last_scheduled_step < max_step : next_update_interval = _call_if_callable ( enabled . update_interval ) next_step = last_scheduled_step + next_update_interval next_delay = _call_if_callable ( enabled . delay ) enabled . update_schedule . append ( ( next_step , next_delay ) ) last_scheduled_step = next_step # Optimize the schedule by planning ahead and dropping unseen entries. enabled . buffer . drop_unobserved_upcoming_items ( enabled . update_schedule , self . _physics_steps_per_control_step )