Автоматизированное получение данных с ip-камер видеонаблюдения с помощью python и requests

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

Channels – действующие каналы (ip-камеры)
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": "" }

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

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

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

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

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

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

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

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

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

Автоматизированное получение данных с ip-камер видеонаблюдения с помощью python и requests

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

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.

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