Безопасность является важной темой в нашей современной жизни, особенно в общественных местах, таких как аэропорты, вокзалы и торговые центры. Одним из распространенных методов обеспечения безопасности является проверка сумок на проходной. Но, как говорится, кто устережёт самих сторожей? Могут ли современные технологии компьютерного зрения наблюдать за охранниками как они за нами? Давайте разберемся!

Навигация:

Подготовка материала с камер видеонаблюдения для последующей работы
Определение людей, сумок и охранников на видео
Определение принадлежности сумки и состояния проверки

Для решения этой проблемы был создан алгоритм детекции проверки охранниками сумок на видеозаписях проходных с использованием YOLO - популярной нейросетевой архитектуры для объектной детекции, в качестве алгоритма трекинга использовался StrongSORT - алгоритм отслеживания объектов между кадрами видео, и алгоритм, основанный на диагоналях найденных рамок объектов для определения проведения осмотра содержимого сумок.

Для тех, кто ~~в танке~~ не знает, что такое YOLO и StrongSORT приведем краткую справку.

YOLO (You Only Look Once) — это однокадровый алгоритм детекции объектов в реальном времени, который основан на глубоком обучении. Алгоритм YOLO использует нейронную сеть, которая способна одновременно обнаруживать несколько объектов на изображении и определять их классы и координаты ограничивающих рамок. Алгоритм YOLO известен своей высокой скоростью работы, что позволяет использовать его в реальном времени на видеозаписях.

StrongSORT — это алгоритм для отслеживания множества объектов, который позволяет отслеживать объекты в видеопотоке в реальном времени. Он основан на использовании оценки оптимального соответствия между обнаруженными объектами на разных кадрах видео и уже существующими отслеживаемыми объектами. Также он предоставляет информацию о скорости и направлении движения объектов.

Отслеживание объектов происходит в два этапа: обновление и прогнозирование. На первом шаге производится обновление состояния объектов на основе данных детекции, полученных от алгоритма YOLO. На втором шаге производится прогнозирование будущего состояния объектов на основе модели движения истории перемещения объектов.

Но перед тем, как приступить к непосредственно к анализу видео, необходимо его подготовить.

Подготовка материала с камер видеонаблюдения для последующей работы

В качестве материала выступали записи с различных камер видеонаблюдения.

Существует множество форматов видеофайлов, используемых в камерах наружного и внутреннего наблюдения. Некоторые из них могут быть специфическими для конкретных моделей камер или производителей, и это составляло основную сложность в подготовке для анализа.

Задача была следующая – перекодировать весь доступный материал с камер к единому знаменателю, в данном случае к формату MP4. На помощь пришёл проект FFmpeg.

FFmpeg — это набор свободных библиотек с открытым исходным кодом, которые позволяют записывать, конвертировать и передавать цифровые аудио и видеозаписи в различных форматах. Для этого были написаны несколько Python-скриптов для автоматизации данного процесса.

Так, например представленный ниже код рекурсивно пробегался по определённой иерархии директорий, в которых находились исходники с камер, и, собственно, запускал в работу FFmpeg, чтобы на выходе получить готовые файлы в формате MP4.

import os 
import subprocess 
from pathlib import Path

input_dir = Path.cwd() / 'input'

files = []

for dir_part, dir_names, file_names in os.walk(input_dir): 
  for file_name in file_names: 
    files.append(os.path.join(dir_part, file_name))

for file in files: 
  ext = file.split('.')[-1]
  if ext == 'mcm':
      output_file = file.replace('mcm', 'mp4')
      print(f'Converting: {file}')
      
      cmd = ('ffmpeg', '-y', '-i', file, '-c', 'copy', output_file)
      subprocess.run(cmd)
  
      print(f'Converted: {output_file}')

Вот кратко как работает данный код:

У нас есть директория input_dir (которая представлена объектом Path), содержащая исходники с камер видеонаблюдения, находящаяся в корневой директории рядом со скриптом.

Проходим через все подкаталоги и файлы в указанном в переменной input_dir и создаем список files, содержащий полные (абсолютные) пути ко всем файлам в этом каталоге и его подкаталогах.

os.walk() используется для рекурсивного обхода дерева каталогов, начиная с корневого каталога input_dir.

Цикл for перебирает каждое имя файла в списке file_names, а затем использует os.path.join() для объединения пути к каталогу dir_part и имени файла file_name в полный (абсолютный) путь к файлу. Этот полный путь добавляется в список files.

Таким образом, после выполнения кода, список files будет содержать примерно следующее:

[
 '/Users/имя_пользователя/Desktop/VideoProject/input/.DS_Store', 
 '/Users/имя_пользователя/Desktop/VideoProject/input/Марата_24.02.23/.DS_Store', 
 '/Users/имя_пользователя/Desktop/VideoProject/input/Марата_24.02.23/24-02-1.mcm', 
 '/Users/имя_пользователя/Desktop/VideoProject/input/Марата_24.02.23/24-02-2.mcm', 
 '/Users/имя_пользователя/Desktop/VideoProject/input/Марата_25.02.23/25-02-1.mcm', 
 '/Users/имя_пользователя/Desktop/VideoProject/input/Марата_25.02.23/25-02-2.mcm'
]

В данном примере нас будут интересовать только файлы с расширением mcm.

MCM — это формат видеофайла, используемый в системах видеонаблюдения от компании Mobotix. Он также является контейнером, содержащим видео и аудио потоки, а также метаданные и другую информацию.

for file in files: 
  ext = file.split('.')[-1]
  if ext == 'mcm':
    output_file = file.replace('mcm', 'mp4')
    print(f'Converting: {file}')
    cmd = ('ffmpeg', '-y', '-i', file, '-c', 'copy', output_file)
    subprocess.run(cmd)
    print(f'Converted: {output_file}')

Данный код проходит через каждый файл в списке files, полученном в предыдущем примере, и проверяет его расширение. Если расширение файла является mcm, то файл конвертируется в mp4 с использованием утилиты командной строки FFmpeg.

Более подробно, каждый путь файла в списке files разбивается на части по символу точки, чтобы получить расширение файла. Полученное расширение записывается в переменную ext. Если расширение равно mcm, то путь к выходному файлу output_file создается с использованием метода replace(), заменяющего расширение mcm на mp4.

Затем создается переменная cmd, содержащая всё необходимое для выполнения конвертации с использованием утилиты командной строки FFmpeg. Эта команда содержит следующие ключи запуска:

FFmpeg — очевидно, вызов утилиты командной строки FFmpeg;
‑y — автоматически подтверждает перезапись выходного файла;
‑i — опция ввода, указывает входной файл;
file — абсолютный путь к имени файла;
‑c — опция кодировки, указывает кодировку выходного файла;
copy — кодек, копирующий поток без перекодировки, ведь всё что нам нужно это изменить контейнер видеофайла с mcm на mp4;
output_file — имя выходного файла.

Затем переменная cmd передаётся в метод subprocess.run() для выполнения. Во время процесса выводится сообщение о конвертации файла и о его успешном завершении.

Таким образом, код выполняет конвертацию всех файлов с расширением mcm в указанном каталоге input_dir и его подкаталогах в файлы с расширением mp4, используя утилиту командной строки FFmpeg.

В данном случае велась работа только с форматом файлов mcm (как один из примеров для наглядности), однако в проекте были и другие форматы с других камер и мест записи, которые хранились c различной иерархией директорий и одним из условий было сохранить данную иерархию хранения.

Например, были и такие материалы, которые представляли один день записи, но не в виде единичного файла (например 25-02-2.mcm), а в виде десятка файлов, которые тем не менее относились к одной сущности – запись за день. Для таких случаев вышеописанный скрипт подстраивался под новые условия. Просто для примера приведу код для одного из таких случаев:

import re

input_dir = Path.cwd() / 'input'

video_dirs = []

regex = re.compile(r'^\d{2}-\d{2}-\d{2}\s\d{2}')

for dir_part, dir_names, file_names in os.walk(input_dir):

  if regex.match(os.path.basename(dir_part)):
    video_dirs.append(dir_part)
    
for dir_part in video_dirs: 
  file_names = os.listdir(dir_part)
  file_paths = [os.path.join(dir_part, file_name) for file_name in file_names]

  cmd = ['ffmpeg', '-i', 'concat:' + '|'.join(file_paths),
         '-crf', '24', f'{dir_part}\output.mp4']

  subprocess.run(cmd)

Данный код выполняет поиск директорий в знакомой нам уже папке input_dir и ее поддиректориях, соответствующих заданному регулярному выражению в переменной regex.

Более конкретно, код создает список video_dirs, куда будут добавляться абсолютные пути к директориям, имена которых начинаются с даты и времени в формате "YY-MM-DD HH" (где YY - год, MM - месяц, DD - день, HH - час). Для этого мы используем функцию os.walk() для рекурсивного обхода всех поддиректорий в input_dir. Когда была найдена директория, имя которой соответствует регулярному выражению regex, ее абсолютный путь добавляется в список video_dirs.

Таким образом, в итоге в списке video_dirs будут содержаться пути к директориям, имена которых начинаются с даты и времени. Этот список будет использовать далее для обработки файлов в найденных директориях.

Интересной здесь так же является следующая строчка кода:

cmd = ['ffmpeg', '-i', 'concat:' + '|'.join(file_paths), 
       '-crf', '24', f'{dir_part}\output.mp4']

Данный код опять же использует утилиту FFmpeg для конкатенации видеофайлов, находящихся по указанным путям file_paths, и сохранения результата в файл с указанным именем в заданной директории dir_part.

Более конкретно, список file_paths содержит абсолютные пути к видеофайлам, которые нужно объединить в один файл. Далее, эти пути объединяются в одну строку с разделителем | (данный разделитель был обусловлен документацией FFmpeg) с помощью метода join(). Затем, формируется список cmd, который содержит имя утилиты FFmpeg, опцию -i для указания входных файлов (соединяемых видеофайлов), опцию -crf для задания качества видео и путь к выходному файлу.

crf 24 — это опция командной строки для утилиты FFmpeg, которая устанавливает значение постоянного качества (constant rate factor, CRF) видео.

CRF является методом сжатия видео, который сохраняет постоянное качество визуального восприятия, независимо от содержания видео, используя переменный битрейт. Это означает, что битрейт может изменяться в зависимости от сложности видео, но общее визуальное качество останется постоянным.

Обычно значения CRF в диапазоне от 18 до 28 используются для достижения хорошего качества при сжатии видео. Однако, для каждого конкретного случая может потребоваться настройка CRF для достижения оптимального баланса между качеством и размером файла, в нашем случае было выбрано значение 24, что на практике показало приемлемое качество видео и размер выходного файла.

В итоге выполнения скрипта ffmpeg склеит указанные видеофайлы в один файл, используя заданное качество видео, и сохранит результирующий файл в указанной директории dir_part с заданным именем output.mp4.

Определение людей, сумок и охранников на видео

Теперь, когда данные подготовлены, перейдем непосредственно к определению охранников и сумок на видео.

Для начала на видео необходимо обнаружить людей, для этого используется функция people_track. Для этого вызываем YOLOv5, где в качестве параметра трекинга указываем StrongSORT и встроенный в нее класс 0, который соответствует людям.

people_track

def people_track(vidos, name): 
  os.system(f'python track.py ‑yolo‑weights yolov5/weights/yolov5m.pt ‑source {vidos} ‑name {name} ‑save‑vid ‑save‑txt ‑classes 0 ‑-tracking‑method strongsort')
  print('Трекинг завершён: ', vidos)

Затем аналогично ищем сумки на видео (функция detect_bags), но указываем встроенные классы 24, 26, 28, которые соответствуют рюкзаку, портфелю и чемодану.

detect_bags

def detect_bags(vidos, conf_thres_yolo, name, path_to_save_hand):
  yolo_run(
    source='runs/track/'+name+'/'+vidos.split('/')[-1],
    weights='yolov5/weights/yolov5m.pt',
    classes=[24, 26, 28],
    conf_thres=conf_thres_yolo,
    nosave=False,
    name=name,
    path_to_save_hand=path_to_save_hand
  )
  print('Детекция звершена: ', vidos)

Однако, нам мало найти только людей, нужно найти еще и охранников, используя get_yolo_security мы обнаруживаем их, дав YOLOv5 веса, которые мы сами обучили находить именно охранников. Алгоритмом get_security агрегируем всех людей по количеству кадров, и из них по перцентилю находим тех, кого больше всего в кадре.

get_yolo_security и get_security

def get_yolo_security(path, persone_id):
    count_detected_security = cls_run(
        'yolov5/weights/guards/yolov5s-cls-guards.pt',
        path_runs+str(persone_id),
        nosave=True
    )
    return count_detected_security


def get_security(path_t_lig, path_runs, ratio_guards):
    man_id = []
    count = []
    count_frame_detect_yolo = []
    people = pd.read_csv(path_to_log, sep=" ", header=None)
    people.columns = ["Frame", "man_id", "x", "y", "w", "h", 
                      "clas", "q2", "q3", "q4","q5"]
    people = people.drop(["q2", "q3", "q4", "q5"], axis=1)
    allPeople = people.groupby("man_id").count().Frame
    allPeople = pd.DataFrame(list(zip(allPeople.index,
                                      allPeople.values)), 
                             coulumns=["man_id","counts"])
    allPeople.to_csv("out_csv/dfO_"+ name + '.csv')
    
    # берём в рассмотрение только тех, 
    # у кого более 10 кадров для невнесения ошибки в перцентиль
    allPeople = allPeople.loc[allPeople['counts'] > 10]
    
    # берём 95 перцентиль, ограничение сверху
    dfOxr = allPeople.loc[allPeople['counts'] > np.percentile(allPeople.counts, 95)]
    dfOxr = dfOxr.man_id.values
    return dfOxr, allPeople

Теперь сохраняем данные о найденных сумках (get_bags_df) в виде Pandas DataFrame. Считываем результаты детекции из файла, разделяем значения по пробелу, добавляем столбцы с координатами и размерами сумок, удаляем ненужные столбцы, и возвращаем полученный DataFrame.

get_bags_df

def get_bags_df(vidos, path_to_save_hand):
    vid = cv2.VideoCapture(vidos)
    heigh_img = vid.get(cv2.CAP_PROP_FRAME_HEIGHT)
    width_img = vid.get(cv2.CAP_PROP_FRAME_WIDTH)
    fps = vid.get(cv2.CAP_PROP_FPS)
    
    # Преобразуем координаты
    bags_df = pd.read_csv(path_to_save_hand, sep=" ", header=None)
    bags_df.max_columns = ['Frame', 'x_t', 'y_t', 'w_t', 'h_t', 
                           'clas', 'temp']
    bags_df['x'] = round((bags_df.x_t - bags_df.w_t/2) * width_img)
    bags_df['y'] = round((bags_df.y_t - bags_df.h_t/2) * heigh_img)
    bags_df['w'] = round((bags_df.w_t) * width_img)
    bags_df['h'] = round((bags_df.w_t) * width_img)
    bags_df['man_id'] = ''
    bags_df = bags_df.drop(['x_t', 'y_t', 'w_t', 'h_t', 'temp'],
                           axis=1)
    return bags_df, fps

Определение принадлежности сумки и состояния проверки

Наконец-то удалось собрать всех людей, сумки и охранников в кадрах, но ведь надо еще найти владельцев сумок и определить, проверяли ли их охранники.

Начинаем с прохода по людям с большими сумками. Для этого используем цикл, который перебирает записи о людях, у которых значение признака "with_big_bag" равно 1, что указывает на наличие большой сумки. Далее производим анализ кадров, на которых были зарегистрированы люди с сумками. Полный код функции довольно обширен, поэтому здесь не приведен, но всегда можно посмотреть полную версию с комментариями на GitHub. А здесь расскажем про общую логику проверки.

Для каждого человека на видеокадрах извлекаем координаты его положения. Проверяем не было ли найдено сумок на этом кадре. Вычисляем диагонали рамок найденных объектов и минимальное расстояние между центрами, если расстояние между центральными точками лежит в пределах (размер человека + размер сумки)/2, то человеку присваивается сумка.

Затем проходим по всем остальным людям на том же видеокадре и ищем охранников. Если находится охранник, то способом, описанном выше, определяем проверял ли он человека с сумкой. Если да, то устанавливаем флаг "checked_bag" в значение 1, что указывает на то, что сумка была проверена охранником на проходной.

В случае, если охранников на видеокадрах нет или были ошибки при извлечении координат рамок, то пропускаем кадр.

Для извлечения координат рамок и расчета расстояний между ними использовался OpenCV, а также NumPy для выполнения операций с массивами чисел.

Путем анализа средних значений диагоналей и минимального расстояния между центрами рамок можно определить, насколько близко подходили охранники к людям с сумками и какие размеры сумок были допущены без проверки.

В итоге мы теперь имеем инструмент, который позволяет нам приводить имеющиеся видео в необходимый формат для yolo и определять насколько качественно охранники проверяют большую ручную кладь. Как думаете, пройдут ли охранники проверку?

Спасибо за ваше внимание!

Сумки в надежных руках благодаря YOLO: проверяем охранников