Время прочтения: 6 мин.

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

Рассмотрим с вам вариант доступа к камерам ТСВ посредством web api на примере api trassir. Доступ к api предоставляется с помощью http get запроса к серверу следующего вида:

https://{ip сервера}:{порт}/{команда}?{параметры выполнения запроса}

В качестве ответа будет получен json, который содержит статус выполнения запроса и сам ответ.

API trassir позволяет получить доступ к real-time видеопотоку и хранилищу с архивом видео путем непосредственного чтения видеофайла (по протоколу TCP/IP, RTSP) или путем генерации скриншотов. Для получения полноценного доступа к api trassir необходимо:

  1. Создать сессию с заданным сервером trassir
  2. Получить перечень каналов (ip-камер), доступных для просмотра на сервере.

Рассмотрим подробнее данные этапы.

Под созданием сессии понимается получение идентификатора sid. Данный идентификатор позволяет пользователю взаимодействовать с сервером trassir через его api. Для того, чтобы создать сессию с сервером, необходимо авторизоваться на сервере выполнив следующий get запрос:

https://{ip сервера}:8080/login?{логин пользователя} =login&password={пароль}

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

Результатом выполнения запроса является json, содержащий 2 параметра:

  1. Success – статус выполнения запроса (1-успешно, 0 – произошла ошибка)
  2. Sid – идентификатор текущей сессии на сервере trassir. Данный идентификатор имеет TTL = 15 минут, который обновляется при каждом совершенном запросе к серверу.

Ниже представлен результат выполнения запроса:

{
	"sid": "CUNKJyO3",
	"success": 1
}

Далее необходимо получить список каналов (ip-камер) для возможности взаимодействия с ними. Для получения перечня каналов выполняется следующий get запрос:

https://{ip сервера}:8080/channels?sid={полученный id сессии на предыдущем шаге}

 В ответе получим json с 2 основными параметрами:

  1. Channels – действующие каналы (ip-камеры)
  2. Zombies – каналы, которые недоступны для просмотра по различным техническим причинам

Для нас интересны только действующие каналы, которые представляют собой массив json объектов следующего вида:

{
	"guid": "{Идентификатор камеры}",
	"name": "{Наименование камеры}",
	"rights": "259",
	"codec": "h264",
	"have_mainstream": "1",
	"have_substream": "1",
	"have_hardware_archive": "0",
	"have_ptz": "0",
	"fish_eye": 0,
	"have_voice_comm": "0",
	"aspect_ratio": "auto",
	"flip": "",
	"rotate": ""
}

В этих объектах нас интересуют поля:

  1. Guid – идентификатор камеры
  2. Name – наименование камеры, задаваемое администратором (необходимо для возможности просмотра фрагмента видео в клиенте trassir)

После получения перечня камер и идентификатора сессии можно полноценно использовать api trassir для получения данных.

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

Для получения скриншота по заданному времени необходимо выполнить следующий get запрос:

https://{ip сервера}:8080/screenshot?sid=sid&guid=guid&timestamp=timestamp}

На выходе мы увидим скриншот с заданной камеры в заданный момент времени.

Вышеперечисленные действия рутинны но достаточно просты, что позволяет автоматизировать их. Далее представлю вам свой вариант автоматизации формирования датасета с изображениями. Для начала определим формат входных данных.

В качестве входных данных используется xlsx файл с таблицей следующего формата:

ipportusernamepasswordcam_namestartend
127.0.0.18080login777101.02.2022 11:0001.02.2022 12:00

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

df = pd.read_csv('1.csv', sep=';')
    queue = mp.Queue()
    servers = {}
    df['start'] = pd.to_datetime(df['start'])
    df['end'] = pd.to_datetime(df['end'])
    for row in df.itertuples():
        key = (row.ip, row.port, row.username, row.password)
        value = (row.start, row.end)
        if (row.ip, row.port, row.username, row.password) not in servers:
            servers[key] = []
        servers[key].append(value)

    for key, value in servers.items():
        queue.put({key: value})

За многопоточную загрузку отвечает метод load класса ProcessingVideo. Данный класс принимает на вход 2 параметра: очередь обработки и количество обработчиков. Внутри данного объекта мы инициализируем объекты Worker, которые в свою очередь создают отдельный поток для загрузки на локальную машину скриншотов посекундно.

class Worker(mp.Process):

	def __init__(self, queue):
		mp.Process.__init__(self)
		self.queue = queue

	def run(self):
		while True:
			if self.queue.empty():
				break
			data = self.queue.get()
			for (ip, port, username, password), operations in data.items():

				trass = RemoteTrassirArchive(ip, port, username, password)
				trass.load_screenshots(operations)
		print('done')


class ProcessingVideo:

	def __init__(self, queue, count_proc=2):
		self.count_proc = count_proc
		self._process_list = []
		self.queue = queue
		self._killer = ProcessKiller(self._process_list)

	def initialize_workers(self):
		for i in range(self.count_proc):
			self._process_list.append(Worker(self.queue))

	def load(self):
		self.initialize_workers()
		try:
			for proc in self._process_list:
				proc.start()
		except KeyboardInterrupt as e:
			print('Аварийное завершение работы')

Управление загрузкой скриншотов определено в объекте RemoteTrassirArchive. Данный объект инициализирует объект Request, который отвечает за выполнение запросов к http серверу trassir и определяет работающие камеры на сервере, инициализируя их в объектах Camera. В объекте Camera реализована основная логика загрузки скриншота с конкретной камеры на сервере trassir. Сама загрузка скриншотов с камеры на конкретном сервере производится последовательно. Решение последовательной загрузки скриншотов с одного сервера было принято для того, чтобы максимально минимизировать нагрузку на канал.

class Request:
    
    def __init__(self, ip, port, sid=''):
        self.ip = ip
        self.port = port
        self.session = sid
        
    def create_session(self, username, password):
        content_type, value = self.get(f'https://{self.ip}:{self.port}/login',params={'username':username, 'password':password})
        self.session = value['sid']     
        
    def _json_deserialize(self, json_text) -> dict:
        return json.loads(json_text[:json_text.index('/*')] if json_text.find('/*')>-1 else json_text)
        
    def get(self, url, params:dict={}):
        if self.session != '':
            params['sid'] = self.session
        response = r.get(url, params=params, verify=False)
        value = self._json_deserialize(response.text) if response.headers['Content-Type'].startswith('application/json') else response.content   
        return response, value

class Camera:
    
    def __init__(self, server, guid, name):
        self._request = server
        self._guid = guid
        a = ':\\/*?<>|"'
        self.name = name
        for char in a:
            self.name = self.name.replace(char, '-')
        
    def _save_image(self, path, img):
            
        with Image(f'{path}.jpeg', 'wb') as f:
            f.write(img)
        
    
    def load_screenshot(self, path: str, timestamp):
        print(path)
        response, img = self._request.get(f'https://{self._request.ip}:{self._request.port}/screenshot/{self._guid}', params={'timestamp': str(timestamp)})
        if not response.headers['Content-Type'].startswith('image/jpeg'):
            print(f'Изображение с камеры {self._guid} по таймкоду {timestamp} не удалось загрузить')
            return
            
        self._save_image(path+'_'+str(timestamp), img)

        
    def load_video(self, dt_start, dt_end):
        dt_s = int(dt_start.timestamp())
        dt_e = int(dt_end.timestamp())
        if not os.path.isdir('data'):
            os.mkdir('data')

        path = f'data/{self._guid}_{self.name}_{dt_s}_{dt_e}'.replace(':', '-')

        for i in range(dt_e-dt_s):
            timestamp = dt_s + i                  
            self.load_screenshot(path, timestamp)
        
        
class RemoteTrassirArchive:
    
    def __init__(self, ip, port, username, password, mock=False):
        self.url = f'https://{ip}:{port}'
        self._server = Request(ip, port)
        self._server.create_session(username, password)
        self.camers={}
        
        response, channels = self._server.get(self.url+'/channels')
        
        self.camers = {}
        
        for channel in channels['channels']:
            self.camers[channel['name']] = Camera(self._server, channel['guid'], channel['name'])
        
    def load_screenshots(self, operations):
        for cam_name, dt_s, dt_e in operations:
            print(cam_name in self.camers.keys())
            if cam_name in self.camers:
                self.camers[cam_name].load_video(dt_s, dt_e)

Сами загруженные картинки записываются в папку data со следующим шаблоном {сервер}_{камера}_{дата начала}_{ дата конца}_{момент времени}.jpeg

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

Спасибо за прочтение данной статьи. Буду рад вашим комментариям.