Последние несколько лет async вообще и asyncio в частности в питоне все больше набирают популярность и их все чаще используют. При этом иногда забывают о принципе KISS (Keep it simple, stupid) и о том, какие вообще проблемы решает асинхронный код и зачем он нужен. В этой статье я бы хотел описать пример, когда задачу можно и, на мой взгляд, нужно решать без использования async. И вообще, практически без всего.
Рассмотрим задачу.
Для начала подключим к компьютеру (ноутбуку) одну вэб-камеру. Далее нам нужно с этой камеры раз в 10 секунд запрашивать кадр и сохранять его. Для простоты примера будем сохранять кадр на диск. Эта задача простая и решение тоже будет простым, даже примитивным я бы сказал.
import time
import cv2
def main():
source = 0
frame_file_path = '<path to a frame file>.jpg'
cap = cv2.VideoCapture(source)
while True:
ret, frame = cap.read()
if not ret:
print('Failed to get a frame')
break
cv2.imwrite(frame_file_path, frame)
time.sleep(10)
cap.release()
if __name__ == '__main__':
main()Чем хорош данный скрипт? Тем, что он прост, понятен и надежен как лом и не требует никаких внешних инструментов типа cron-а.
Теперь немного масштабируем наш пример: добавим еще одну камеру и скажем, что с нее нам нужны кадры каждые 15 секунд. Если на проекте вы активно пользуетесь asyncio, то можете по инерции поступить следующим образом: сказать, что раз задача по существу I/O Bound, а камер стало больше, то это работа для asyncio. Затем взять какой-нибудь планировщик задач типа APScheduler, который поддерживает asyncio и с его помощью и помощью лома накидать что-то вроде такого.
import asyncio
from functools import partial
import cv2
from apscheduler.schedulers.asyncio import AsyncIOScheduler
async def job(source, target_file_path):
cap = cv2.VideoCapture(source)
ret, frame = await asyncio.get_event_loop().run_in_executor(None, partial(cap.read, cap))
if not ret:
print('Failed to get a frame')
return
await asyncio.get_event_loop().run_in_executor(None, partial(cv2.imwrite, target_file_path, frame))
def main():
frame_0_file_path = ''
frame_1_file_path = ''
scheduler = AsyncIOScheduler()
scheduler.add_job(job, 'interval', seconds=10, kwargs={'source': 0, 'target_file_path': frame_0_file_path})
scheduler.add_job(job, 'interval', seconds=15, kwargs={'source': 1, 'target_file_path': frame_1_file_path})
scheduler.start()
try:
asyncio.get_event_loop().run_forever()
except (KeyboardInterrupt, SystemExit):
pass
if __name__ == '__main__':
main()На первый взгляд все неплохо, но это на первый взгляд. Помимо дополнительной внешней зависимости мы собрали под капотом корутины, asyncio с ивент лупом и потоки для запуска синхронных функций асинхронным образом. И теперь вместо топорной и надежной программки у нас есть целый технический букет, с которым можно повозиться и от этого кайфануть. Но оно нам надо? Мне вот как инженеру не очень :) Если вспомнить, что async, ивент луп и вот это все призваны при написании I/O Bound приложений экономить ресурсы при большой нагрузке, а нагрузки у нас нет, да и камер будет конечное и вполне осмысливаемое число, то могут закрасться смутные сомнения. А нужен ли нам там asyncio тут вообще? И можно прийти к выводу, что не нужен и что вместо запуска асинхронных задач на ивент лупе, можно воспользоваться классом apscheduler.executors.pool.ProcessPoolExecutor, который позволит нам запускать наши джобы в процессах. Есть еще класс, который позволяет нам запускать джобы в потоках. Но сомнения у нас смутные, поэтому мы его тоже отбросим. Получится что-то вроде такого:
import cv2
from apscheduler.schedulers.blocking import BlockingScheduler
def job(source, target_file_path):
cap = cv2.VideoCapture(source)
ret, frame = cap.read()
if not ret:
print('Failed to get a frame')
return
cv2.imwrite(target_file_path, frame)
def main():
frame_0_file_path = ''
frame_1_file_path = ''
scheduler = BlockingScheduler({
'apscheduler.executors.default': {
'class': 'apscheduler.executors.pool:ProcessPoolExecutor',
'max_workers': 2
}
})
scheduler.add_job(job, 'interval', seconds=10, kwargs={'source': 0, 'target_file_path': frame_0_file_path})
scheduler.add_job(job, 'interval', seconds=15, kwargs={'source': 1, 'target_file_path': frame_1_file_path})
scheduler.start()
if __name__ == '__main__':
main()
Так намного лучше: нет потоков, нет asyncio с ивент лупом не по делу. Но все еще не идеал, ибо есть APScheduler с какими-то внутренним механизмами. Оно нам надо? Мне - нет. Наше решение для одной камеры было максимально простым, тупым, надежным и имело всего одну зависимость от которой, увы, не убежишь. Почему бы нам тогда не вспомнить про принцип KISS еще разок и не взять первую версию программы, не параметризовать ее и не запускать ее для каждой камеры? Этот подход мне лично нравится больше всего: минимум усилий (если подумать о нем сразу), максимум надежности (я как инженер люблю все надежное) и максимально просто сопровождение, а точнее его отсутствие :)
В итоге получаем следующее:
import argparse
import time
import cv2
def main(source: int, frame_file_path: str, interval: int):
cap = cv2.VideoCapture(source)
while True:
ret, frame = cap.read()
if not ret:
print('Failed to get a frame')
break
cv2.imwrite(frame_file_path, frame)
time.sleep(interval)
cap.release()
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--source', type=int)
parser.add_argument('--target-frame-path', type=str, dest='target_frame_path')
parser.add_argument('--interval', type=int)
args = parser.parse_args()
main(args.source, args.target_frame_path, args.interval)Такое же простое, надежное и понятное решение как лом. А против лома, как известно, приема нет :)
Комментарии (16)
OkunevPY
07.06.2022 21:13Сколько холивара на пустом месте, посыл автора очень простой, не усложняйте то что можно оставить простым. Где тут было хоть слово про правильность приведённых примеров как истины последней инстанции? Это же просто примеры, они могут быть как хорошие так и плохие по коду или оформлению, но должны решать свою задачу, и тут они с задачей справились.
Реализаций может бесчисленно множество, вплоть до реализации своего мини крона. Но тут вопрос зачем это всё? Где тут усмотрели требования к точности? Или требования к незапуску второго процесса пока не закончился первый? Откуда вообще взялось столько домыслоф о которых автор нигде даже косвенно не упоминал?
SharUpOff
08.06.2022 00:16+1async, ивент луп и вот это все призваны при написании I/O Bound приложений экономить ресурсы при большой нагрузке
Не совсем так. Асинхронный код всего лишь позволяет не ждать I/O в mainloop, вместо этого продолжая его последовательное (!) выполнение. Нельзя асинхронно производить вычисления - для этого нужна многопоточность. Только с точки зрения I/O - наверное, да, можно сказать про экономию ресурсов. Один асинхронный процесс обработает больше входных данных за период времени из-за отсутствия задержек на ожидание ввода. При этом, большое количество возвратов в mainloop и использование async функций там, где нет непосредственно I/O, создаёт оверхед и увеличивает нагрузку на процессор.
gecube
ага, и тут мы выясняем, что, оказывается, запуск по timer'у (крону) это совершенно не то же самое, что и слип. Крон дает гарантирую запуска в момент времени X (скажем, каждую 5 минуту в 00 секунду), плюс-минус какую-то погрешность. А слип? А слип это ТОТ самый код, который, во-первых, в неудачной реализации жрет 100% cpu, а, во-вторых, не гарантирует ничего. У Вас погрешность запуска будет расти с каждым последующим запуском. И если изначально таймаут для слипа был подобран для 5 минут, то потому Вы внезапно увидите, что задержка запуска составляет 1 час. Я уж не говорю - о всяких приключениях вроде снижения частоты процессора из-за энергосбережения, приколы ntp, или попросту перевод часов... и обо всем этом $&$#*(&$#( приходится думать. С уважением...
Moraiatw
Где это вы такой Sleep взяли? В Windows:
Спит то количество миллисекунд, которое ей указано, при этом нагружает процессор на 0%, т.к. потоку просто не выделяется процессорное время.
Может в питоне какой то свой слип...
gecube
реализация на уровне ЯП, насколько мне известно, не дает каких-либо гарантий. Да, основные реализации могут не жечь такты cpu, но тут уж как карта ляжет :-)
И всегда стоит задавать дополнительные вопросы "а что если?"
AtachiShadow
Дело в том, что сам скрипт исполняется какое-то время, допустим 5 секунд, а после этого включается слип на 10 секунд, в итоге получаем периодичность не 10 секунд как хотели а 10+5. Из-за этого таймер должен считаться независимо от исполнения основного скрипта, и тут опять же или вешать потоки/корутины/процессы или просто убрать из скрипта контроль периода и отдать его штатному инструменту автозапуска
ZyXI
Ещё можно просто использовать
. Штатный инструмент автозапуска, это, конечно, хорошо, но, во‐первых, с ним вы каждый раз будете платить за запуск процесса Python, чтение всех модулей, …, во‐вторых, останавливать скрипт выше гораздо удобнее. В варианте выше период должен плавать не так сильно.
Moraiatw
Вобщем я понял претензию к слипу, но тут есть 2 стороны. Все зависит от задачи.
Если есть задача выполнять скрипт строго в определенные отметки времени, допустим, запустить процесс бекапа в 03:00 - то да, это лучше делать кроном.
Если задача - запускать какое то действие с паузами в 10 сек, то лучше Sleep (здорового человека, а не который 100% CPU). Время выполнения действия может быть в общем случае неопределено. При использовании крона (или другого средства планирования) может возникнуть ситуация, когда предыдущее действие еще не завершилось, а следующее уже запускается, со всеми вытекающими.
gecube
эта проблема легко решается. В баш скриптах, например, используется утилита flock. В python - аналогично. Что-то такого плана https://gist.github.com/jirihnidek/430d45c54311661b47fb45a3a7846537
Moraiatw
Т.е. как я понимаю, скрипт будет ждать завершения предыдущего инстанса? И в итоге, если время выполнения действия больше интервала запуска, то ждущие инстансы будут накапливаться до исчерпания ресурсов системы?
gecube
доку читали? Там есть разные варианты поведения - можно ждать освобождения блокировки, а можно падать с ошибкой, если ее не взять (именно второе поведение позволяет не плодить ждущие инстансы). Задается разница буквально одной опцией...
Moraiatw
Доку не читал.
Звучит как то не очень отказоустойчиво...
gecube
ну, обработайте ошибку ) от задачи же зависит.
Moraiatw
Тогда мы получим пропуски срабатываний ) это все не годится.
AtachiShadow
Да, идея автора понятна как тот самый лом и естественно правильна, не нужно переусложнять простые вещи, но действительно реализацию автора можно улучшить. Вы предлагаете использовать крон, а я предложу на системах с systemd использовать таймеры в юнитах. Там у них целый раздел [Timer] есть, где можно настраивать "каждые 2 минуты" или "каждый вторник в 12:00" или другие варианты...
Так или иначе какая-то служба автозапуска будет использована для запуска самого скрипта, почему бы не использовать тогда штатные средства этих самых систем автозапуска. Даже в Планировщике задач Windows есть настройки периодичности
gecube
Добрый вечер! Да, естественно, если речь идет про crond vs systemd-timer, я при прочих равных выберу последний. В своем же сообщении "крон" я подразумевал как некий принцип некоего планировщика - не важно, будь он на уровне ЯП в виде какой-то подключаемой библиотеки, или как-то еще реализован, но никак не внешний запускатор (хотя и он возможен - при учете, например, контроля за временем выполнения скрипта, чтобы не копились "мертвые" или "зависшие" процессы в системе). То есть в своем сообщении я не имел в виду крон как системный демон. Прошу прощения, если ввел в заблуждение...