Модуль sched предоставляет гибкий и удобный способ для планирования задач, которые должны быть выполнены в определённое время. Этот модуль идеально подходит для ситуаций, когда необходимо выполнять действия с задержкой или через заданные интервалы.
Модуль sched основан на концепции планировщика (scheduler), который управляет очередью событий. Каждое событие представляет собой вызов функции с заданными аргументами. Планировщик использует временную функцию (time function) для отслеживания текущего времени и функцию задержки (delay function) для ожидания перед выполнением следующего события.
Основной класс в модуле — это scheduler. Он определяет универсальный интерфейс для планирования событий. Его конструктор принимает две функции:
class sched.scheduler(timefunc=time.monotonic, delayfunc=time.sleep)
timefunc: функция, возвращающая текущее время. Функция должна вызываться без аргументов и возвращать число ("время" в любых единицах измерения). По умолчанию в качестве временной функции используется time.monotonic(), что делает класс
устойчивым к изменениям системного времени. Это особенно полезно для длительных процессов.
delayfunc: функция, которая приостанавливает выполнение на заданное количество секунд. Должна вызываться с одним аргументом, совместима с выводом timefunc и должна задерживать выполнение на указанное количество единиц времени. delayfunc также будет вызываться с аргументом 0 после каждого события, чтобы дать другим потокам возможность работать в многопоточных приложениях.
Пример определения объекта класса:
import sched import time s = sched.scheduler(time.time, time.sleep)
В этом примере используется стандартная связка time.time и time.sleep, которая подходит для большинства простых случаев.
Итак, планировщик scheduler управляет очередью событий - вызываемые функции, соответственно нам нужно определить события. Для добавления события в очередь используются два метода:
enter(delay, priority, action, arguments, kwargs)
Планирует событие на основе задержки и принимает следующие параметры:
delay: задержка, через которую должно быть выполнено событие.
priority: приоритет события (число). События с меньшим числом выполняются раньше, если их время выполнения совпадает.
action: вызываемая функция.
arguments: кортеж позиционных аргументов для функции.
kwargs: словарь именованных аргументов.
Фактически выполнение события означает выполнение вызова action(*arguments, **kwargs).
enterabs(time, priority, action, arguments, kwargs): планирует событие на основе абсолютного времени.
Параметр time представляет время, когда должно быть выполнено событие. Этот параметр должен иметь числовой тип, совместимый с возвращаемым значением функции timefunc, переданной конструктору.
Остальные параметры те же, что и у enter().
Залпанированные события нам надо запустить. Для этого необходима функция run():
scheduler.run(blocking=True)
Эта функция будет ожидать (используя функцию delayfunc, переданную конструктору) следующего события, затем выполнит его и так далее, пока не останется запланированных событий.
Рассмотрим простой пример с выполнением одного события (одной функции):
import sched
import time
def print_message():
print("Hello METANIT.COM!")
# создаем scheduler
s = sched.scheduler(time.time, time.sleep)
print("Начало планирования...")
# планируем выполнение через 3 минуты
s.enter(3, 1, print_message)
# запускаем выполнение
s.run()
print("Планирование завершено.")
В этом примере вызов s.enter() устанавливает выполнение действия - функции print_message через 3 минуты, а вызов s.run() запускает цикл планировщика,
который ожидает и выполняет событие - функции print_message.
Возьмем другой, чуть более сложный пример, где нужно вывести несколько сообщений с задержкой:
import sched
import time
s = sched.scheduler(time.time, time.sleep)
def print_message(message):
print(f"[{time.strftime("%H:%M:%S", time.localtime())}] - {message}")
print("Начало планирования...")
s.enter(3, 1, print_message, ("Первое сообщение (через 3 секунды)",))
s.enter(5, 1, print_message, ("Второе сообщение (через 5 секунд)",))
s.enter(5, 2, print_message, ("Третье сообщение (через 5 секунд, но с более низким приоритетом)",))
# Событие с приоритетом 1 будет выполнено раньше, чем событие с приоритетом 2, если время совпадает.
s.run()
print("Планирование завершено.")
Ожидаемый вывод:
Начало планирования... [07:20:03] - Первое сообщение (через 3 секунды) [07:20:05] - Второе сообщение (через 5 секунд) [07:20:05] - Третье сообщение (через 5 секунд, но с более низким приоритетом) Планирование завершено.
Модуль sched сам по себе не предоставляет встроенного механизма для повторяющихся задач. Однако, это легко реализовать, планируя следующее событие внутри самой вызываемой функции:
import sched
import time
s = sched.scheduler(time.time, time.sleep)
def repeat_task():
print(f"[{time.strftime("%H:%M:%S", time.localtime())}] - Hello METANIT.COM")
# Планируем следующий вызов этой же функции
s.enter(3, 1, repeat_task)
print("Начало повторяющейся задачи...")
s.enter(3, 1, repeat_task)
s.run()
В этом примере repeat_task вызывает сама себя, обеспечивая непрерывное выполнение с заданным интервалом - 3 минуты. С помощью стандартных механизмов операционной системы -
комбинации клавиш Ctrl+Z/Ctrl+C мы можем завершить выполнение планировщика и программы.
В качестве альтернативы можно использовать цикл:
import sched
import time
s = sched.scheduler(time.time, time.sleep)
def repeat_task():
print(f"[{time.strftime("%H:%M:%S", time.localtime())}] - Hello METANIT.COM")
print("Начало повторяющейся задачи...")
while True:
s.enter(3, 1, repeat_task)
s.run()
Можно еще тоньше настроить выполнение, например, выполнить через некоторый промежуток времени определенное число раз (например, 5 раз):
import sched
import time
s = sched.scheduler(time.time, time.sleep)
def repeat_task():
print(f"[{time.strftime("%H:%M:%S", time.localtime())}] - Hello METANIT.COM")
print("Начало повторяющейся задачи...")
count = 5 # 5 раз выполняем задачу
while count:
s.enter(3, 1, repeat_task)
s.run()
count -= 1
События можно отменить, используя метод cancel(event). Метод enter или enterabs возвращает объект события, который можно использовать для отмены:
import sched
import time
s = sched.scheduler(time.time, time.sleep)
def my_task():
print("Эта задача будет отменена.")
def cancel_task(event):
s.cancel(event)
print("Задача успешно отменена.")
event_to_cancel = s.enter(5, 1, my_task)
s.enter(2, 1, cancel_task, (event_to_cancel,))
s.run()
В этом примере my_task не будет выполнена, так как cancel_task отменит её до наступления времени выполнения.
Преимущества:
Простота: Очень простой и интуитивно понятный интерфейс.
Изолированность: Работает в рамках одного потока, не требует сложной синхронизации.
Гибкость: Позволяет легко планировать события с различной задержкой и приоритетом.
Устойчивость к сбоям времени: По умолчанию использует time.monotonic(), что делает его надёжным для долгосрочных задач.
Недостатки:
Блокировка: Метод run() блокирует текущий поток до выполнения всех запланированных событий. Это делает его непригодным для асинхронных приложений.
Точность: Точность выполнения зависит от функции задержки (time.sleep) и может быть не идеальной для задач, требующих высокой точности.
Ограниченная функциональность: Не имеет встроенных возможностей для многопоточности или обработки ошибок, в отличие от более сложных фреймворков.
