1

Хочу реализовать в цикле для каждого условия таймер. То есть

While True:
if *** == 'first choice':
    # Do something
    time.sleep(5)
if *** == 'second choice':
    # Do something
    time.sleep(5)

Но при этом, чтобы эти таймеры работали асинхронно, где-то в фоне. Чтобы я мог сделать им запрос *.is_running() и, если получу True, то оборвать цикл. Как такое вообще реализовать?

С asyncio я поломал себе всю голову и мало что понял, а реализовать это костыльно через одни if-ы - как-то "непрофессионально" и некрасиво. Если то, что я описал, невозможно, то какие еще есть варианты?

3 ответа 3

1
import asyncio, random

async def check(name, func):
    while True:
        if func():  # проверка func условия
            print(name, 'break')
            break  # и, если получу True, то оборвать цикл
        print(name, 'continue')
        await asyncio.sleep(.5)

async def append_var(var, var_min_max):
    while True:
        i = random.randint(*var_min_max)
        var.append(i)  # создает какие-то условия (var) для дальнейшей проверки таймерами
        print('Append Var >>>', i)
        await asyncio.sleep(.5)

async def main(var):
    asyncio.create_task(append_var(var, var_min_max=(1, 25)))  # условия
    checks = [
        lambda: var[-1] in [10, 20, 21],
        lambda: var[-1] % 2,
        lambda: len(var) > 2 and var[-2] in [5, 15, 25]
    ]  # для каждого условия таймер
    # чтобы эти таймеры работали асинхронно, где-то в фоне
    tasks = [asyncio.create_task(check(f'check_{n}', f)) for n, f in enumerate(checks, start=1)]
    await asyncio.gather(*tasks)  # ждать выполнения всех условий
    print('exit')

if __name__ == '__main__':
    var = [0]
    asyncio.run(main(var))

out:

Append Var >>> 6
check_1 continue
check_2 continue
check_3 continue
Append Var >>> 6
check_2 continue
check_1 continue
check_3 continue
Append Var >>> 1
check_1 continue
check_2 break
check_3 continue
Append Var >>> 15
check_1 continue
check_3 continue
Append Var >>> 12
check_1 continue
check_3 break
Append Var >>> 24
check_1 continue
Append Var >>> 21
check_1 break
exit
0

Вы можете использовать модуль threading.

import threading
import time

def timer():
    time.sleep(5)
    print("test")

def start_submit_thread(event):
    global submit_thread
    submit_thread = threading.Thread(target=timer)
    submit_thread.daemon = True
    submit_thread.start()
    check_submit_thread()

def check_submit_thread():
    while True:
        if submit_thread.is_alive():
           time.sleep(0.2)
           check_submit_thread()
        if stop_threads:
           break   
        else:
          print("Timer stopped")

def stop_upload_thread():
    if submit_thread.is_alive():
       global stop_threads 
       stop_threads = True
       print("Thread stopped")
    else:
       print("something occured")   

start_submit_thread(None)    

В функции timer() вы можете вставить ваш таймер, в функции submit_thread.is_alive() программа проверяет работает ли timer(), чтобы остановить цикл можете использовать stop_upload_thread(). Чтобы начать цикл можете использовать start_submit_thread() а чтобы проверить работает ли таймер def check_submit_thread().

0

ИМХО, в твоем случае, проще реализовать таймер. Вот пример:

from datetime import datetime, timedelta
from time import sleep

class Timer:
    def __init__(self, seconds):
        self._delta = timedelta(seconds=seconds)
        self._stop = datetime.now() + self._delta
    def is_running(self):
        return datetime.now() < self._stop

t = Timer(5)
while True:
    if not t.is_running():
        break
    print('ok')
    sleep(1)  # Это чтоб много не печатало.

Ваш ответ

Нажимая «Отправить ответ», вы соглашаетесь с условиями пользования и подтверждаете, что прочитали политику конфиденциальности.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.