Привет, Хабр!
Подготовительным этапом для видеоаналитики с применением методов машинного обучения - является выгрузка записей из видеорегистраторов Hikvision, что является достаточно длительной задачей, особенно если регистраторов несколько, давайте попытаемся разобраться как выполнить эту работу быстрее и удобнее, используя сервер.
Основанием для разработки «Task-Сервера» - послужила задача по выгрузке видеозаписей с видеорегистраторов Hikvision. В работе была использована библиотека Hikload которая подключалась к регистраторам и загружала видеоролики.
Для начала немного о «Task-Сервере» (Рис. 1), в данном случае это небольшой сервер, хранящий в базе данных или файле заранее подготовленные строки, которые будут переданы в качестве параметров клиентам, подключенным к хосту. Клиент же в свою очередь запускает скрипт с полученными данными и после выполнения запрашивает новую задачу до тех пор, пока задачи не закончатся.
В качестве инструмента для создания такого сервера был выбран микрофреймворк Flask, нам достаточно небольшого веб-приложения для нашей задачи. Из плюсов Flask можно выделить простоту создания веб-приложений, отсутствие сложных настроек и достаточно стабильная и быстрая работа. В качестве хранилища для команд была выбрана таблица Excel. Все задачи были сформированы заранее.
Проблема
Так как количество камер в разных помещениях, разных городов достаточно много, а время выгрузки с одного устройства сильно варьировалось, от нескольких минут до часа (Рис. 2), то было принято решение: на основе параметров для подключения к регистраторам сформировать задачи и распределить их на разных машинах, для чего и потребовался сервер.
Решение:
Начнем работу с создания сервера, для начала создаем новое окружение под приложение следующей командой в терминале:
python –m venv venvТак как решение было использовать Flask то подключаем следующие модули:
«Flask» - для создания веб-приложения.
«openpyxl» – непосредственно для работы с excel файлом, где находятся команды.
«multiprocessing» - из этого модуля нам потребуются только очереди, так как они идеально подходят для нашей задачи.
Для начала импортируем библиотеку Flask и инициализируем приложение:
from flask import Flask
app = Flask(__name__)Импортируем остальные библиотеки:
from openpyxl import load_workbook
from multiprocessing import QueueСоздаем и наполняем базу данных задачами, которые сервер будет возвращать клиенту. Задачи сформированы заранее, они состоят из сервера/ ip – адреса к которому подключаемся, user - логина, password - пароля, cameras - номера камеры. По умолчанию если не задавать дату и время - то Hikload выгружает видео за один текущий день. Полный список параметров для выгрузки видео можно узнать, набрав «hikload --help» (рис. 3)
Далее добавим функцию, которая возвращает команду из базы данных. Все задачи из таблицы Excel выгружаются в Queue(очередь).
data = load_workbook('test.xlsx', 'wb')
sheet = data.active
que = Queue()
result = []
#выгрузка задач из таблицы Excel и запись в очередь
for row in sheet:
for cell in row:
result.append(cell.value)
que.put(result)
result = []
#выдача команд из очереди
def get_task():
return que.get()
#отправка команд клиентам.Декоратор «app.route» регистрирует URL-адрес, по которому будет доступно представление, описанное ниже и вызывается экземпляром приложения app(Flask):
@app.route('/')
@app.route('/index')
def index():
return ' '.join(map(str, get_task()))Весь код сервера представлен ниже:
from flask import Flask
app = Flask(__name__)
from openpyxl import load_workbook
from multiprocessing import Queue
data = load_workbook('test.xlsx', 'wb')
sheet = data.active
que = Queue()
result = []
#выгрузка задач из таблицы Excel и запись в очередь
for row in sheet:
for cell in row:
result.append(cell.value)
que.put(result)
result = []
#выдача команд из очереди
def get_task():
return que.get()
#отправка команд клиентам
@app.route('/')
@app.route('/index')
def index():
if not que.empty():
return ' '.join(map(str, get_task()))
else:
return
if __name__ == '__main__':
app.run(debug=False)Сервер готов, очередь заполнена задачами, следующим пунктом будет разработка клиента для правильной интерпретации принятых команд и запуска соответствующих скриптов. Рецепт программы-клиента состоит из следующих ингредиентов:
«Requests» для обращения к серверу и получения задачи.
def get_task():
for _ in range(2):
resp = rq.get('localhost')
if resp != “the end”:
tasks.append(list((resp.content.decode('utf-8').split(' '))))
else:
return None
return (tasks)«Multiprocessing» для выполнения 2-х задач (можно и другое количество, зависящее от характеристик вашего устройства) параллельно.
if __name__ == '__main__':
while True:
tasks = [] #создаем пустой список для задач
response = get_task()
NUM_CORE = 2 #объявляем количество ядер
read = mp.Queue()
[read.put(x) for x in response] #записываем очередь командами
print("read qsize", read.qsize())
for i in range(NUM_CORE):
mp.Process(target=load, args=(read,)).start() #запускаем скрипт
mp.join()
while True:
time.sleep(1)Модуль «os» непосредственно, для запуска Hikload.
def load(read):
while True:
if not read.empty(): #проверяем заполненность очереди
x = read.get()
name_proc = mp.current_process().name
print(f"{name_proc} stared") #выводим работающие процессы
task = (f'hikload --server={x[2]} '
f'--user={x[0]} '
f'--password={x[1]} '
f'--cameras={x[3]} '
f'--downloads {x[2]+ "_" + name_proc}'
)
print(task)
'''Запуск скрипта с нашими командами'''
os.system(task)
else:
breakКод клиента целиком выглядит следующим образом:
import multiprocessing as mp, os, time, requests as rq
#получаем задачи с сервера
def get_task():
for _ in range(2):
resp = rq.get('http://127.0.0.1:5000/')
if resp != 'the end':
tasks.append(list((resp.content.decode('utf-8').split(' '))))
else:
return None
return (tasks)
#запуск выгрузки видеозаписей
def load(read):
while True:
if not read.empty(): #проверяем заполненность очереди
x = read.get()
name_proc = mp.current_process().name
print(f"{name_proc} stared") #выводим работающие процессы
task = (f'hikload --server={x[2]} '
f'--user={x[0]} '
f'--password={x[1]} '
f'--cameras={x[3] '
f'--downloads {x[2]+ "_" + name_proc}'
)
print(task)
'''Запуск скрипта с нашими командами'''
os.system(task)
else:
break
if __name__ == '__main__':
while True:
tasks = []
response = get_task()
NUM_CORE = 2 #объявляем количество ядер
read = mp.Queue()
get_task() #вызываем функцию загрузки задач
[read.put(x) for x in tasks] #записываем очередь командами
print("read qsize", read.qsize())
for i in range(NUM_CORE):
mp.Process(target=load, args=(read,)).start() #запускаем скрипт
mp.join()
while True:
time.sleep(1)Для начала запускаем сервер (Рис. 5) на локальном хосте, следующей командой: flask run
Далее запускаем клиент, команда: python main.py:
Мы видим, что очередь автоматически заполнилась 2-мя задачами (Рис. 6) и клиент начал выгрузку, в это время сервер показывает, что передал 2 команды, код 200 в ответе говорит об успешной обработке запроса (Рис. 7)
Под каждую команду создаются отдельные папки, где хранятся выгруженные видеозаписи, названия формируются параметром --download (рис.8). После отработки/выгрузки команды – клиент снова посылает запрос на сервер и получает новую команду.
И, конечно, вывод
В результате мы реализовали автоматизированное выполнение задачи, разделенной на множество подзадач. На каждом клиенте выполняются 2 выгрузки параллельно (рис.9)
Из явных преимуществ такого подхода можно выделить:
Загрузка видеозаписей проходит быстрее с каждым подключенным клиентом.
Все задачи распределяются автоматически.
Сервер прост в разработке, многофункционален и его можно использовать для любых, похожих на показанной в примере задач.
