def _update()

in ax/utils/testing/backend_simulator.py [0:0]

• 15 lines of code
• 10 McCabe index (conditional complexity)

    def _update(self, timestamp: float) -> None:
        """Check if trials have completed (or stopped) and update the simulator.

        Args:
            timestamp: The current timestamp.
        """

        completed_since_last = self._update_completed(timestamp)

        # if no trial has finished since the last call we're done
        if len(completed_since_last) == 0:
            return

        # if at least one trial has finished, we need to graduate queued trials to
        # running trials. Since all we need to keep track of is the start_time, we can
        # do this retroactively.
        # TODO: Improve performance / make less ad hoc by using a priority queue
        for c in completed_since_last:
            if self.num_queued > 0:
                new_running_trial = self._queued.pop(0)
                sim_start_time = (
                    # pyre-fixme[58]: `+` is not supported for operand types
                    #  `Optional[float]` and `float`.
                    c.sim_start_time + c.sim_runtime
                    if not self.use_update_as_start_time
                    else self.time
                )
                new_running_trial.sim_start_time = sim_start_time
                self._running.append(new_running_trial)

        # since these graduated trials could both have started and finished in between
        # the simulation updates, we need to re-run the update with the new state
        self._update(timestamp)