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

    This function determines which of the proposed upcoming observations will
    never in fact be delivered and removes them from the observation schedule.

    We assume that observations will only be queried at times that are integer
    multiples of `read_interval`. If more observations are generated during
    the upcoming control step than the `buffer_size` of this `Buffer`
    then of those new observations will never be required. This function takes
    into account the delayed arrival time and existing buffered items in the
    planning process.

    Args:
      observation_schedule: An list of `(timestamp, delay)` tuples, where
        `timestamp` is the time at which the observation value will be produced,
        and `delay` is the amount of time the observation will be delayed by.
        This list will be modified in place.
      read_interval: The time interval between successive calls to `read`.
        We assume that observations will only be queried at times that are
        integer multiples of `read_interval`.
Plans an optimal observation schedule for an upcoming control period.
Plans an optimal observation schedule for an upcoming control period .
def drop_unobserved_upcoming_items(self, observation_schedule, read_interval):
    """Plans an optimal observation schedule for an upcoming control period.

    This function determines which of the proposed upcoming observations will
    never in fact be delivered and removes them from the observation schedule.

    We assume that observations will only be queried at times that are integer
    multiples of `read_interval`. If more observations are generated during
    the upcoming control step than the `buffer_size` of this `Buffer`
    then of those new observations will never be required. This function takes
    into account the delayed arrival time and existing buffered items in the
    planning process.

    Args:
      observation_schedule: An list of `(timestamp, delay)` tuples, where
        `timestamp` is the time at which the observation value will be produced,
        and `delay` is the amount of time the observation will be delayed by.
        This list will be modified in place.
      read_interval: The time interval between successive calls to `read`.
        We assume that observations will only be queried at times that are
        integer multiples of `read_interval`.
    """
    # Private deques to simulate what the deques will look like in the future,
    # according to the proposed upcoming observation schedule.
    future_arrived_deque = collections.deque()
    future_pending_deque = collections.deque()

    # Take existing buffered observations into account when planning the
    # upcoming schedule.
    def get_next_existing_timestamp():
      for obs in reversed(self._pending_deque):
        yield InFlightObservation(obs.timestamp, obs.delay, None)
      while True:
        yield InFlightObservation(-np.inf, 0, None)
    existing_timestamp_iter = get_next_existing_timestamp()
    existing_timestamp = next(existing_timestamp_iter)

    # Build the simulated state of the pending deque at the end of the proposed
    # schedule.
    sorted_schedule = sorted([InFlightObservation(time[0], time[1], None)
                              for time in observation_schedule])
    for new_timestamp in reversed(sorted_schedule):
      # We don't need to worry about any existing item that are delivered before
      # the first new item, since those are purged independently of our
      # proposed new observations.
      while existing_timestamp.arrival > new_timestamp.arrival:
        future_pending_deque.appendleft(existing_timestamp)
        existing_timestamp = next(existing_timestamp_iter)
      future_pending_deque.appendleft(new_timestamp)

    # Find the next timestep at which `read` is called.
    first_proposed_timestamp = min(t for t, _ in observation_schedule)
    next_read_time = read_interval * int(np.ceil(
        first_proposed_timestamp // read_interval))

    # Build the simulated state of the arrived deque at each subsequent
    # control steps.
    while future_pending_deque:
      # Keep track of observations that are delivered for the first time
      # during this control timestep.
      newly_arrived = collections.deque()
      while (future_pending_deque and
             future_pending_deque[0].arrival <= next_read_time):
        # `fake_observation` is an `InFlightObservation` without `value`.
        fake_observation = future_pending_deque.popleft()
        future_arrived_deque.append(fake_observation)
        newly_arrived.append(fake_observation)
      while len(future_arrived_deque) > self._buffer_size:
        stale = future_arrived_deque.popleft()
        # Newly-arrived items that become immediately stale are never actually
        # delivered.
        if newly_arrived and stale == newly_arrived[0]:
          newly_arrived.popleft()
          # `stale` might either be one of the existing pending observations or
          # from the proposed schedule.
          if stale.timestamp >= first_proposed_timestamp:
            observation_schedule.remove((stale.timestamp, stale.delay))

      next_read_time += read_interval
def drop_unobserved_upcoming_items ( self , observation_schedule , read_interval ) : # Private deques to simulate what the deques will look like in the future, # according to the proposed upcoming observation schedule. future_arrived_deque = collections . deque ( ) future_pending_deque = collections . deque ( ) # Take existing buffered observations into account when planning the # upcoming schedule. def get_next_existing_timestamp ( ) : for obs in reversed ( self . _pending_deque ) : yield InFlightObservation ( obs . timestamp , obs . delay , None ) while True : yield InFlightObservation ( - np . inf , 0 , None ) existing_timestamp_iter = get_next_existing_timestamp ( ) existing_timestamp = next ( existing_timestamp_iter ) # Build the simulated state of the pending deque at the end of the proposed # schedule. sorted_schedule = sorted ( [ InFlightObservation ( time [ 0 ] , time [ 1 ] , None ) for time in observation_schedule ] ) for new_timestamp in reversed ( sorted_schedule ) : # We don't need to worry about any existing item that are delivered before # the first new item, since those are purged independently of our # proposed new observations. while existing_timestamp . arrival > new_timestamp . arrival : future_pending_deque . appendleft ( existing_timestamp ) existing_timestamp = next ( existing_timestamp_iter ) future_pending_deque . appendleft ( new_timestamp ) # Find the next timestep at which `read` is called. first_proposed_timestamp = min ( t for t , _ in observation_schedule ) next_read_time = read_interval * int ( np . ceil ( first_proposed_timestamp // read_interval ) ) # Build the simulated state of the arrived deque at each subsequent # control steps. while future_pending_deque : # Keep track of observations that are delivered for the first time # during this control timestep. newly_arrived = collections . deque ( ) while ( future_pending_deque and future_pending_deque [ 0 ] . arrival <= next_read_time ) : # `fake_observation` is an `InFlightObservation` without `value`. fake_observation = future_pending_deque . popleft ( ) future_arrived_deque . append ( fake_observation ) newly_arrived . append ( fake_observation ) while len ( future_arrived_deque ) > self . _buffer_size : stale = future_arrived_deque . popleft ( ) # Newly-arrived items that become immediately stale are never actually # delivered. if newly_arrived and stale == newly_arrived [ 0 ] : newly_arrived . popleft ( ) # `stale` might either be one of the existing pending observations or # from the proposed schedule. if stale . timestamp >= first_proposed_timestamp : observation_schedule . remove ( ( stale . timestamp , stale . delay ) ) next_read_time += read_interval