Есть ли в Python простой способ дождаться выполнения определенного условия?

Question

Мне нужно подождать в сценарии, пока не выполнится определенное количество условий?

Я знаю, что могу создавать свои собственные события, используя условные переменные и друзей, но я не хочу проходить через все трудности, связанные с его реализацией, поскольку некоторые изменения свойств объекта происходят из внешнего потока в обернутой библиотеке C ++ (Boost. Python), поэтому я не могу просто перехватить __setattr__ в классе и поместить туда условную переменную, что оставляет мне либо попытку создать и передать переменную условия Python из C ++, либо обернуть собственную переменную и ждать ее в Python, оба из которых звучат странно, излишне сложно и скучно.

Есть ли более простой способ сделать это, исключая непрерывный опрос условия?

В идеале это было бы по линии

res = wait_until(lambda: some_predicate, timeout)
if (not res):
    print 'timed out'

Answer 1

К сожалению, единственная возможность удовлетворить ваши ограничения - периодически опрос , например ....:

import time

def wait_until(somepredicate, timeout, period=0.25, *args, **kwargs):
  mustend = time.time() + timeout
  while time.time() < mustend:
    if somepredicate(*args, **kwargs): return True
    time.sleep(period)
  return False

или тому подобное. Это можно оптимизировать несколькими способами, если somepredicate можно разложить (например, если известно, что оно является and из нескольких предложений, особенно если некоторые из предложений, в свою очередь, подлежат оптимизации путем обнаружения с помощью threading.Event s или что угодно, и т. д. и т. д.), но в общих чертах, которые вы просите, этот неэффективный подход - единственный выход.

Answer 2

Вы в основном ответили на свой вопрос: нет.

Поскольку вы имеете дело с внешними библиотеками в boost.python, которые могут изменять объекты в свободное время, вам нужно либо заставить эти подпрограммы вызывать обновление обработчика событий, либо работать с условием.

Answer 3

Вот расширение потоков к решению Алекса:

import time
import threading

# based on https://stackoverflow.com/a/2785908/1056345                                                                                                                                                                                                                                                                         
def wait_until(somepredicate, timeout, period=0.25, *args, **kwargs):
    must_end = time.time() + timeout
    while time.time() < must_end:
        if somepredicate(*args, **kwargs):
            return True
        time.sleep(period)
    return False

def wait_until_par(*args, **kwargs):
    t = threading.Thread(target=wait_until, args=args, kwargs=kwargs)
    t.start()
    print ('wait_until_par exits, thread runs in background')

def test():
    print('test')

wait_until_par(test, 5)

Answer 4

Предлагаемое решение:

def wait_until(delegate, timeout: int):
end = time.time() + timeout
while time.time() < end:
    if delegate():
        return True
    else:
        time.sleep(0.1)
return False

Использование:

wait_until(lambda: True, 2)

Answer 5

Это сработало для меня

direction = ''
    t = 0
    while direction == '' and t <= 1:
        sleep(0.1)
        t += 0.1

Это для ожидания сигнала с определением времени в 1 секунду