Есть ли простой способ в Python подождать, пока какое-то условие будет истинным?
Мне нужно ждать в script до тех пор, пока не вернется определенное количество условий?
Я знаю, что могу качать мои собственные события с использованием переменных условий и друзей, но я не хочу испытывать проблемы с его реализацией, поскольку некоторые изменения свойств объекта происходят из внешнего потока в завернутой библиотеке С++ (Boost. Python), поэтому я не могу просто захватить __setattr__
в классе и поставить там переменную условия, которая оставляет меня с попыткой создать и сигнализировать переменную состояния Python из С++, или обернуть родной, и ждать на ней в Python, оба из которых звучат неудобно, бесполезно сложны и скучны.
Есть ли более простой способ сделать это, запрещая непрерывный опрос условия?
В идеале это было бы вдоль линий
res = wait_until(lambda: some_predicate, timeout)
if (not res):
print 'timed out'
Ответы
Ответ 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
или любого другого, и т.д.), но в общих чертах, о которых вы просите, этот неэффективный подход является единственным выходом.
Ответ 2
Вы в основном ответили на свой вопрос: нет.
Поскольку вы имеете дело с внешними библиотеками в boost.python, которые могут изменять объекты на досуге, вам нужно либо включить эти процедуры, либо обновить обработчик событий, либо работать с условием.
Ответ 3
Вот расширение для решения Alex:
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)
Ответ 4
Это сработало для меня
direction = ''
t = 0
while direction == '' and t <= 1:
sleep(0.1)
t += 0.1
Это для ожидания сигнала с определением времени в 1 секунду.
Ответ 5
Предложенное решение:
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)