Отслеживание бездействия Python

Добрый день! Подскажите, пожалуйста, как можно реализовать выполнение функции на Python в консольном приложении по таймауту?
Смысл в следующем: Есть консольное приложение, которое идет своим чередом и периодически требует от пользователя каких-либо действий. В случае же, если пользователь проявляет бездействие в течение определенного времени, необходимо выполнить определенную функцию.
Из того, что есть: Функция для отслеживания времени (Пример):
StartTime = time.time() // В теле программе отслеживаем время последнего действия TimeOut = 2000 def countTime(): CurTime = time.time() // текущее время If CurTime > (StartTime+TimeOut): // проверяем разницу между текущим временем и последним действием, сравниваем с таймаутом print SomeFunction() // выполняем необходимую функцию
Но я не имею никаких предположений, как запускать функцию проверки времени параллельно (асинхронно?) выполнению основного алгоритма программы.
Заранее спасибо!

Ответ

если пользователь проявляет бездействие в течение определенного времени, необходимо выполнить определенную функцию.
Общий подход: запланировать выполнение функции на определённое время и при любой активности пользователя переносить запуск на более позднее время.
Чтобы просто запланировать выполнение функции func через 5 секунд в Питоне, не блокируя основной поток:
import threading
timer = threading.Timer(5, func) timer.start()
Если timer.cancel() (отмена) не вызвать, то через 5 секунд будет вызвана func() в фоновом потоке.
Чтобы адаптировать для вашей задачи, необходимо метод restart() добавить, который запустит отсчёт заново, при любой активности пользователя. Этот подход достаточно распространён, чтобы своё имя заработать: Watchdog timer — букв. «сторожевой пёс»
watchdog = WatchdogTimer(timeout, func, daemon=True) watchdog.start() # ... здесь программа что-то делает # и перезапускает отсчёт при любой активности пользователя watchdog.restart()
# в конце отменяем вызов func совсем или просто из программы выходим watchdog.cancel()
где WatchdogTimer класс определён здесь
Вот пример кода, где Питон-скрипт завершается, если пользователь не активен больше заданного времени в X (на Unix)

Необязательно потоки использовать, например, в программе, использующей asyncio
async def watchdog(loop, last_activity_time, timeout, func, *args): "Run *func(*args)* if more than *timeout* seconds since *last_activity_time*." while (loop.time() - last_activity_time()) < timeout: await asyncio.sleep(1) return func(*args)
Цель данной реализации—простота: пока разница между текущим временем (loop.time()) и записанным временем последней активности пользователя (last_activity_time()) меньше заданного значения (timeout), спим. В противном случае вызываем заданную функцию (func). Можно придумать решение, которое не использует asyncio.sleep(1)
Пример использования:
#!/usr/bin/env python3 import asyncio import random
loop = asyncio.get_event_loop() user = {'last_activity': loop.time()} asyncio.ensure_future(watchdog(loop, lambda: user['last_activity'], 5, print, 'idle')) async def simulate_activity(): for i in range(20): print(i) if random.random() < .1: user['last_activity'] = loop.time() # user have done something now await asyncio.sleep(1) loop.run_until_complete(simulate_activity())
Программа печатает idle, если пользователь, имитированный с помощью random в simulate_activity() coroutine, слишком долго не показывает активности.