waiter.py 1.8 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950
  1. """
  2. Copyright (c) Contributors to the Open 3D Engine Project.
  3. For complete copyright and license terms please see the LICENSE at the root of this distribution.
  4. SPDX-License-Identifier: Apache-2.0 OR MIT
  5. Functions that repeatedly run until a condition is met
  6. """
  7. import time
  8. def wait_for(fn_act, timeout=30, exc=None, interval=1):
  9. """
  10. Continues to execute a given function until the function returns True. Raises an exception if the function does
  11. not return True before the timeout
  12. :param fn_act: The target function to execute
  13. :param timeout: The set amount of time before raising an exception
  14. :param exc: The exception to raise. An assertion error is raised by default
  15. :param interval: The time to wait between subsequent function calls
  16. """
  17. timeout_end = time.time() + timeout
  18. while not fn_act():
  19. if time.time() > timeout_end:
  20. if exc is not None:
  21. raise exc
  22. else:
  23. assert False, 'Timeout waiting for {}() after {}s.'.format(fn_act.__name__, timeout)
  24. time.sleep(interval)
  25. def wait_while(fn_act, timeout, exc=None, interval=1):
  26. """
  27. Continues to execute a given function until the specified amount of time has passed. Raises an exception if the
  28. function does not return True during this time.
  29. :param fn_act: The target function to execute
  30. :param timeout: The set amount of time to wait
  31. :param exc: The exception to raise. An assertion error is raised by default
  32. :param interval: The time to wait between subsequent function calls
  33. """
  34. timeout_end = time.time() + timeout
  35. while time.time() < timeout_end:
  36. if not fn_act():
  37. if exc is not None:
  38. raise exc
  39. else:
  40. assert False, '{}() failed while waiting for {}s'.format(fn_act.__name__, timeout)
  41. time.sleep(interval)