- Создал(а) Рудковский Михаил (доб. 1236), редактировал(а) Дмитрий Гаврилов июл 03, 2024
resources/archive/shot_saver.py
Модуль предоставляет высокоуровневый интерфейс для асинхронного сохранения скриншотов в виде класса ShotSaver.
Аргументы класса
pool_size (int, optional)
: Максимальное кол-во одновременно сохраняемых скриншотов. По умолчанию10
.callback (callable[str, str], optional)
: Callback функция, которая вызывается после сохранения скриншотов. В функцию передается guid задачи на сохранение и текущий статус задачи.
Публичные атрибуты класса
timeout_sec (int)
: Время ожидания скриншота, сек. По умолчанию10
.buffer_ts(int)
: Время ожидания для создания скриншота с фигурами. По умолчанию 0.max_tries (int)
: Кол-во попыток сохранить скриншот. По умолчанию2
.callback (callable[str, str])
: Тоже что аргументcallback
pool_size (int)
: Тоже что аргументpool_size
screenshots_folder (str)
: Папка сохранения скриншотов по умолчанию если не задана в другая при вызове метода. По умолчанию папка скриншота сервера.
Методы класса
save
- Сохранение скриншота в файл. Возвращает guid задачи и путь, куда скриншот будет сохранен.
Аргументы метода:channel_full_guid (str)
: Guid канала видаChannelGuid_ServerGuid
dt (datetime.datetime, optional)
: Дата/время сохранения скриншота. По умолчанию текущая дата/время.figures (bool, optional)
: ЕслиTrue
- сохраняет скриншот с фигурами. По умолчаниюFalse
.file_name (str, optional)
: Название файла скриншота. По умолчанию формируется по шаблону{channel.name} (%Y.%m.%d %H-%M-%S.%f).jpg
В имени скриншота поддерживается форматирование даты/времени (доступные аргументы доступны на сайте https://strftime.org/) а также объектchannel
иserver
.file_path (str, optional)
: Полный путь для сохранения скриншота. По умолчанию сохраняет в папку, которая задана в атрибутеscreenshots_folder
callback (callable[str, str], optional)
: Callback функция, которая вызывается после сохранения скриншотов. В функцию передается guid задачи на сохранение и текущий статус задачи. Если не задана - используетсяcallback
класса.task_guid (str, optional)
: Guid задачи. По умолчанию задается с помощью функцииhost.random_guid()
get
- Возвращает вcallback
изображение, bytes
Аргументы метода:channel_full_guid (str)
: Guid канала видаChannelGuid_ServerGuid
dt (datetime.datetime, optional)
: Дата/время сохранения скриншота. По умолчанию текущая дата/время.figures (bool, optional)
: ЕслиTrue
- сохраняет скриншот с фигурами. По умолчаниюFalse
.callback (callable[str, str])
: Callback функция, которая вызывается после сохранения скриншотов. В функцию передается guid задачи на сохранение и текущий статус задачи. Если не задана - используетсяcallback
класса. Если еще иcallback
класса не задана, вызовет ошибкуEnvironmentError
Возможные статусы сохранения скриншота содержатся в объекте status
:
status.saving
- Скриншот сохраняется.status.error
- Ошибка сохранения скриншота.status.success
- Скриншот успешно сохранен.status.in_queue
- Скриншот добавлен в очередь на сохранение.
При инициализации класс ShotSaver
создает thread pool из нескольких отдельных потоков (worker
), кол-во потоков равно количеству задач, но не более чем pool_size, с помощью метода create_workers Класса ThreadPool.
Сохранение скриншотов в thread pool необходимо для ограничения кол-ва одновременно запрашиваемых скриншотов, т.к. большое кол-во запросов метода host.screenshot_v2
или host.screenshot_v2_figures
может привести к падению трассира.
Каждый вызов метода save
добавляет задачу на сохранение в очередь задач. При этом проверяется количество потоков (worker), если задач больше чем потоков, будут созданы дополнительные потоки, макс. количество потоков равно pool_size. Впоследствии задачи забирают worker'ы по очереди. А также в callback передается status.in_queue.
Worker вызывает метод host.screenshot_v2
или host.screenshot_v2_figures
в зависимости от параметра figures и передает в callback
status.saving
. После чего ожидает в течении заданного в параметре timeout_sec
времени появление файла в указанной папке.
Если скриншот не появился в течении заданного времени в папке и max_tries > 1
то worker
будет вызовывать метод сохранения скриншота снова. Далее логика ожидания и повторного сохранения повторяется столько раз, сколько задано в параметра max_tries
.
Если скриншот успешно сохранился, то в callback
передается status.success
.
Если скриншот не появился в папке после всех попыток в callback передается status.error
.
После окончания работы, если очередь пустая, worker заканчивает свою работу самостоятельно, при поступлении новых задач будут созданы новые worker'ы.
Метод get
работает также как метод save
, за исключением того, что после успешного сохранения скриншота он считывает изображение и полученный результат передает в callback, после чего удаляет файл скриншота.
Примеры кода
Простой пример
Просто сохраняем текущий скриншот с канала со стандартным именем в папку скриншотов.
from shot_saver import ShotSaver CHANNEL_GUID = "uR76TlcA_O4Yazs8u" # Необходимо заменть на guid Вашего канала ss = ShotSaver() ss.save(CHANNEL_GUID)
Изменение папки сохранения скриншотов
Если Вы хотите, чтобы все скриншоты данного скрипта сохранялись в отдельную папку, можно изменить ее глобально для всего класса.
from shot_saver import ShotSaver CHANNEL_GUID = "uR76TlcA_O4Yazs8u" # Необходимо заменть на guid Вашего канала ss = ShotSaver() ss.screenshots_folder = r"D:\DSSL\shots" ss.save(CHANNEL_GUID)
Или задать непосредственно при сохранении скриншота
from shot_saver import ShotSaver CHANNEL_GUID = "uR76TlcA_O4Yazs8u" # Необходимо заменть на guid Вашего канала ss = ShotSaver() ss.save(CHANNEL_GUID, file_path=r"D:\DSSL\shots")
Запрос скриншота за определенную дату/время
Также мы можем менять и имя скриншота, и/или добавлять дату/время за которое хотим сохранить скриншот.
Для примера сохраним скриншот за текущую дату, а время изменим на 00:00:00
import datetime from shot_saver import ShotSaver CHANNEL_GUID = "uR76TlcA_O4Yazs8u" # Необходимо заменть на guid Вашего канала ss = ShotSaver() dt = datetime.datetime.combine( datetime.datetime.now().date(), # Текущая дата datetime.time() # Время 00:00:00 ) ss.save(CHANNEL_GUID, dt=dt, file_name="midnight_shot.jpg")
Отслеживание статуса сохранения скриншота
При успешном сохранении открытие полученного изображения.
Мы используем переменную tasks
для хранения текущих задач и путей сохранения скриншота для каждой задачи.
import os import host from shot_saver import ShotSaver, status CHANNEL_GUID = "uR76TlcA_O4Yazs8u" ss = ShotSaver() tasks = {} def callback(guid, state): host.message("[%s] %s" % (guid, state)) # Отбражаем текущий статус задачи if state == status.success: # Открываем изображение если скриншот успешно сохранился fpath = tasks.pop(guid) os.startfile(fpath) elif state == status.error: # Просто удаляем задачу из списка, если произошла ошибка fpath = tasks.pop(guid) host.message("Can't save shot: %s" % fpath) task_guid, file_path = ss.save(CHANNEL_GUID, callback=callback) tasks[task_guid] = file_path
Получение изображения в bytes
Например для отображения в окне host.question
import base64 import host from shot_saver import ShotSaver CHANNEL_GUID = "uR76TlcA_O4Yazs8u" ss = ShotSaver() __image_html = """<img src="data:image/png;base64,{img}" width="800">""" def show_image(image_bytes): host.question( __image_html.format(img=base64.b64encode(image_bytes)), "OK", lambda: 0 ) ss.get(CHANNEL_GUID, callback=show_image)