1. Цели
Обзор
В этой кодовой лаборатории основное внимание будет уделено созданию комплексного приложения Vertex AI Vision для мониторинга размера очереди с использованием розничных видеоматериалов . Мы будем использовать предварительно обученные встроенные функции аналитики занятости специализированной модели, чтобы фиксировать следующие вещи:
- Посчитайте количество людей, стоящих в очереди.
- Подсчитайте количество людей, обслуженных у стойки.
Что вы узнаете
- Как создать приложение в Vertex AI Vision и развернуть его
- Как настроить поток RTSP с использованием видеофайла и вставить поток в Vertex AI Vision с помощью vaictl из Jupyter Notebook.
- Как использовать модель анализа занятости и ее различные функции.
- Как искать видео в вашем хранилище Медиа-хранилище Vertex AI Vision.
- Как подключить выходные данные к BigQuery, написать SQL-запрос для извлечения информации из выходных данных модели в формате JSON и использовать выходные данные для маркировки и аннотирования исходного видео.
Расходы:
Общая стоимость запуска этой лаборатории в Google Cloud составляет около 2 долларов США.
2. Прежде чем начать
Создайте проект и включите API:
- В консоли Google Cloud на странице выбора проекта выберите или создайте проект Google Cloud . Примечание . Если вы не планируете сохранять ресурсы, созданные в ходе этой процедуры, создайте проект вместо выбора существующего проекта. После выполнения этих шагов вы можете удалить проект, удалив все ресурсы, связанные с ним. Перейти к выбору проектов
- Убедитесь, что для вашего облачного проекта включена оплата. Узнайте, как проверить, включена ли оплата в проекте .
- Включите Compute Engine, Vertex API, Notebook API и Vision AI API. Включите API
Создайте учетную запись службы:
- В консоли Google Cloud перейдите на страницу «Создать учетную запись службы» . Перейти к созданию учетной записи службы.
- Выберите свой проект.
- В поле Имя учетной записи службы введите имя. Консоль Google Cloud заполняет поле «Идентификатор учетной записи службы» на основе этого имени. В поле Описание учетной записи службы введите описание. Например, учетная запись службы для быстрого запуска.
- Нажмите «Создать» и продолжите .
- Чтобы предоставить доступ к вашему проекту, предоставьте следующую роль своему сервисному аккаунту:
- Vision AI > Редактор Vision AI
- Compute Engine > Администрирование экземпляра Compute (бета-версия)
- BigQuery > Администратор BigQuery.
В списке Выберите роль выберите роль. Для дополнительных ролей нажмите «Добавить еще одну роль» и добавьте каждую дополнительную роль.
- Нажмите Продолжить .
- Нажмите «Готово» , чтобы завершить создание учетной записи службы. Не закрывайте окно браузера. Вы будете использовать его на следующем шаге.
3. Настройте блокнот Jupyter.
Прежде чем создавать приложение в Occupancy Analytics, вы должны зарегистрировать поток, который может использоваться приложением позже.
В этом руководстве вы создадите экземпляр Jupyter Notebook, на котором будет размещаться видео, и отправите эти потоковые видеоданные из блокнота. Мы используем блокнот Jupyter, поскольку он дает нам возможность выполнять команды оболочки, а также запускать собственный код предварительной и последующей обработки в одном месте, что очень удобно для быстрого экспериментирования. Мы будем использовать этот блокнот, чтобы:
- Запустить rtsp- сервер в качестве фонового процесса
- Запустите команду vaictl в качестве фонового процесса.
- Запускайте запросы и обрабатывайте код для анализа результатов анализа занятости.
Создайте блокнот Jupyter
Первым шагом в отправке видео из экземпляра Jupyter Notebook является создание записной книжки с нашей учетной записью службы, созданной на предыдущем шаге.
- В консоли перейдите на страницу Vertex AI. Перейдите в Vertex AI Workbench.
- Нажмите «Блокноты, управляемые пользователями».
- Нажмите «Новый ноутбук» > Tensorflow Enterprise 2.6 (с LTS) > «Без графических процессоров».
- Введите имя блокнота Jupyter. Дополнительные сведения см. в разделе Соглашение об именовании ресурсов .
- Нажмите «ДОПОЛНИТЕЛЬНЫЕ ОПЦИИ».
- Прокрутите вниз до разделов разрешений.
- Снимите флажок «Использовать учетную запись службы Compute Engine по умолчанию».
- Добавьте адрес электронной почты учетной записи службы, созданный на предыдущем шаге. И нажмите «Создать».
- После создания экземпляра нажмите ОТКРЫТЬ JUPYTERLAB.
4. Настройте ноутбук для потоковой передачи видео.
Прежде чем создавать приложение в Occupancy Analytics, вы должны зарегистрировать поток, который может использоваться приложением позже.
В этом руководстве мы будем использовать наш экземпляр Jupyter Notebook для размещения видео, а вы отправите эти потоковые видеоданные с терминала Notebook.
Загрузите инструмент командной строки vaictl
- В открытом экземпляре Jupyterlab откройте блокнот из панели запуска.
- Загрузите инструмент командной строки Vertex AI Vision (vaictl), инструмент командной строки сервера rtsp, инструмент open-cv, используя следующую команду в ячейке записной книжки:
!wget -q https://github.com/aler9/rtsp-simple-server/releases/download/v0.20.4/rtsp-simple-server_v0.20.4_linux_amd64.tar.gz
!wget -q https://github.com/google/visionai/releases/download/v0.0.4/visionai_0.0-4_amd64.deb
!tar -xf rtsp-simple-server_v0.20.4_linux_amd64.tar.gz
!pip install opencv-python --quiet
!sudo apt-get -qq remove -y visionai
!sudo apt-get -qq install -y ./visionai_0.0-4_amd64.deb
!sudo apt-get -qq install -y ffmpeg
5. Загрузите видеофайл для потоковой передачи.
После настройки среды ноутбука с помощью необходимых инструментов командной строки вы можете скопировать образец видеофайла, а затем использовать vaictl для потоковой передачи видеоданных в приложение анализа занятости.
Зарегистрируйте новый поток
- Нажмите вкладку «Потоки» на левой панели Vertex AI Vision.
- Нажмите кнопку «Зарегистрироваться» вверху.
- В имени потока введите «queue-stream».
- В разделе «Регион» выберите тот же регион, который был выбран при создании блокнота на предыдущем шаге.
- Нажмите Зарегистрироваться
Скопируйте образец видео на свою виртуальную машину.
- Скопируйте в свой блокнот образец видео с помощью следующей команды wget.
!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4
Потоковое видео с виртуальной машины и вставка данных в свой поток
- Чтобы отправить этот локальный видеофайл во входной поток приложения, используйте следующую команду в ячейке записной книжки. Вы должны сделать следующие замены переменных:
- PROJECT_ID: идентификатор вашего проекта Google Cloud.
- МЕСТОПОЛОЖЕНИЕ: идентификатор вашего местоположения. Например, us-central1. Дополнительную информацию см. в разделе Облачные местоположения .
- LOCAL_FILE: имя локального видеофайла. Например,
seq25_h264.
PROJECT_ID='<Your Google Cloud project ID>'
LOCATION='<Your stream location>'
LOCAL_FILE='seq25_h264.mp4'
STREAM_NAME='queue-stream'
- Запускаем rtsp-simple-server, на котором мы транслируем видеофайл по протоколу rtsp.
import os
import time
import subprocess
subprocess.Popen(["nohup", "./rtsp-simple-server"], stdout=open('rtsp_out.log', 'a'), stderr=open('rtsp_err.log', 'a'), preexec_fn=os.setpgrp)
time.sleep(5)
- Используйте инструмент командной строки ffmpeg для зацикливания видео в потоке rtsp.
subprocess.Popen(["nohup", "ffmpeg", "-re", "-stream_loop", "-1", "-i", LOCAL_FILE, "-c", "copy", "-f", "rtsp", f"rtsp://localhost:8554/{LOCAL_FILE.split('.')[0]}"], stdout=open('ffmpeg_out.log', 'a'), stderr=open('ffmpeg_err.log', 'a'), preexec_fn=os.setpgrp)
time.sleep(5)
- Используйте инструмент командной строки vaictl для потоковой передачи видео с uri rtsp-сервера в поток очереди Vertex AI Vision, созданный на предыдущем шаге.
subprocess.Popen(["nohup", "vaictl", "-p", PROJECT_ID, "-l", LOCATION, "-c", "application-cluster-0", "--service-endpoint", "visionai.googleapis.com", "send", "rtsp", "to", "streams", "queue-stream", "--rtsp-uri", f"rtsp://localhost:8554/{LOCAL_FILE.split('.')[0]}"], stdout=open('vaictl_out.log', 'a'), stderr=open('vaictl_err.log', 'a'), preexec_fn=os.setpgrp)
Между запуском операции загрузки vaictl и появлением видео на информационной панели может пройти около 100 секунд.
После того, как прием потока станет доступен, вы сможете просмотреть видеопоток на вкладке «Потоки » панели управления Vertex AI Vision, выбрав поток потока очереди.
6. Создайте заявку
Первый шаг — создать приложение, которое обрабатывает ваши данные. Приложение можно рассматривать как автоматизированный конвейер, который соединяет следующее:
- Прием данных : видеопоток вводится в поток.
- Анализ данных : после приема можно добавить модель искусственного интеллекта (компьютерного зрения).
- Хранение данных : две версии видеопотока (исходный поток и поток, обработанный моделью искусственного интеллекта) могут храниться в хранилище мультимедиа.
В консоли Google Cloud приложение представлено в виде графика.
Создать пустое приложение
Прежде чем вы сможете заполнить граф приложения, вы должны сначала создать пустое приложение.
Создайте приложение в консоли Google Cloud.
- Перейдите в консоль Google Cloud.
- Откройте вкладку «Приложения» панели управления Vertex AI Vision. Перейдите на вкладку «Приложения»
- Нажмите кнопку «Создать» .
- Введите « queue-app» в качестве имени приложения и выберите свой регион.
- Нажмите Создать .
Добавьте узлы компонентов приложения
После создания пустого приложения вы можете добавить три узла в граф приложения:
- Узел приема : ресурс потока, который принимает данные, отправленные с видеосервера rtsp, созданного вами в блокноте.
- Узел обработки : модель анализа занятости, которая действует на основе полученных данных.
- Узел хранения : медиа-хранилище, в котором хранятся обработанные видео и служит хранилищем метаданных. Хранилища метаданных включают аналитическую информацию о загруженных видеоданных и информацию, полученную с помощью моделей искусственного интеллекта.
Добавьте узлы компонентов в свое приложение в консоли.
- Откройте вкладку «Приложения» панели управления Vertex AI Vision. Перейдите на вкладку «Приложения»
Это приведет вас к графической визуализации конвейера обработки.
Добавьте узел приема данных
- Чтобы добавить узел входного потока, выберите параметр «Потоки» в разделе «Соединители» бокового меню.
- В разделе «Источник» открывшегося меню «Поток» выберите «Добавить потоки» .
- В меню «Добавить потоки» выберите «queue-stream» .
- Чтобы добавить поток в график приложения, нажмите «Добавить потоки» .
Добавляем узел обработки данных
- Чтобы добавить узел модели подсчета занятости, выберите параметр «Аналитика занятости» в разделе «Специализированные модели» бокового меню.
- Оставьте настройки по умолчанию Люди . Снимите флажок «Транспортные средства» , если он уже выбран.
- В разделе «Дополнительные параметры» нажмите « Создать активные зоны/линии».
- Нарисуйте активные зоны с помощью инструмента «Многоугольник» для подсчета людей в этой зоне. Пометьте зону соответствующим образом
- Нажмите на стрелку назад вверху.
- Добавьте настройки времени ожидания для обнаружения перегрузки, установив флажок.
Добавить узел хранения данных
- Чтобы добавить узел назначения вывода (хранилища), выберите параметр «Vision AI Warehouse» в разделе «Соединители» бокового меню.
- Нажмите на Vertex AI Warehouse Connector, чтобы открыть его меню, нажмите «Подключить склад» .
- В меню Подключить склад выберите Создать новый склад . Назовите складскую очередь-склад и оставьте продолжительность TTL 14 дней.
- Нажмите кнопку «Создать» , чтобы добавить склад.
7. Подключите вывод к таблице BigQuery.
Когда вы добавляете коннектор BigQuery в приложение Vertex AI Vision, все выходные данные модели подключенного приложения будут загружены в целевую таблицу.
Вы можете либо создать собственную таблицу BigQuery и указать эту таблицу при добавлении соединителя BigQuery в приложение, либо позволить платформе приложения Vertex AI Vision автоматически создать таблицу.
Автоматическое создание таблиц
Если вы разрешите платформе приложений Vertex AI Vision автоматически создавать таблицу, вы можете указать этот параметр при добавлении узла соединителя BigQuery.
Если вы хотите использовать автоматическое создание таблиц, применяются следующие условия набора данных и таблиц:
- Набор данных: имя автоматически создаваемого набора данных — Visionai_dataset.
- Таблица. Имя автоматически создаваемой таблицы — Visionai_dataset.APPLICATION_ID.
- Обработка ошибок:
- Если таблица с таким же именем в том же наборе данных существует, автоматического создания не происходит.
- Откройте вкладку «Приложения» панели управления Vertex AI Vision. Перейдите на вкладку «Приложения»
- Выберите «Просмотреть приложение» рядом с названием вашего приложения из списка.
- На странице конструктора приложений выберите BigQuery в разделе Соединители .
- Оставьте поле пути к BigQuery пустым.
- В метаданных магазина: выберите только «Аналитика занятости» и снимите флажки с потоков.
Окончательный график приложения должен выглядеть так:
8. Разверните свое приложение для использования
После того как вы создали комплексное приложение со всеми необходимыми компонентами, последним шагом к использованию приложения является его развертывание.
- Откройте вкладку «Приложения» панели управления Vertex AI Vision. Перейдите на вкладку «Приложения»
- Выберите «Просмотреть приложение» рядом с приложением -очередью в списке.
- На странице Studio нажмите кнопку «Развернуть» .
- В следующем диалоговом окне подтверждения нажмите « Развернуть» . Операция развертывания может занять несколько минут. После завершения развертывания рядом с узлами появятся зеленые галочки.
9. Поиск видеоконтента на складе хранения
После загрузки видеоданных в приложение для обработки вы можете просмотреть проанализированные видеоданные и выполнить поиск данных на основе информации анализа занятости.
- Откройте вкладку «Склады» панели управления Vertex AI Vision. Перейдите на вкладку «Склады».
- Найдите в списке склад-очередь и нажмите « Просмотреть активы» .
- В разделе «Количество людей» установите минимальное значение 1 и максимальное значение 5.
- Чтобы отфильтровать обработанные видеоданные, хранящиеся в медиа-хранилище Vertex AI Vision, нажмите «Поиск» .
Просмотр сохраненных видеоданных, соответствующих критериям поиска в консоли Google Cloud.
10. Аннотируйте и анализируйте выходные данные с помощью таблицы BigQuery.
- В блокноте инициализируйте следующие переменные в ячейке.
DATASET_ID='vision_ai_dataset'
bq_table=f'{PROJECT_ID}.{DATASET_ID}.queue-app'
frame_buffer_size=10000
frame_buffer_error_milliseconds=5
dashboard_update_delay_seconds=3
rtsp_url='rtsp://localhost:8554/seq25_h264'
- Теперь мы будем захватывать кадры из потока rtsp, используя следующий код:
import cv2
import threading
from collections import OrderedDict
from datetime import datetime, timezone
frame_buffer = OrderedDict()
frame_buffer_lock = threading.Lock()
stream = cv2.VideoCapture(rtsp_url)
def read_frames(stream):
global frames
while True:
ret, frame = stream.read()
frame_ts = datetime.now(timezone.utc).timestamp() * 1000
if ret:
with frame_buffer_lock:
while len(frame_buffer) >= frame_buffer_size:
_ = frame_buffer.popitem(last=False)
frame_buffer[frame_ts] = frame
frame_buffer_thread = threading.Thread(target=read_frames, args=(stream,))
frame_buffer_thread.start()
print('Waiting for stream initialization')
while not list(frame_buffer.keys()): pass
print('Stream Initialized')
- Извлеките временную метку данных и информацию аннотаций из таблицы bigquery и создайте каталог для хранения захваченных изображений кадров:
from google.cloud import bigquery
import pandas as pd
client = bigquery.Client(project=PROJECT_ID)
query = f"""
SELECT MAX(ingestion_time) AS ts
FROM `{bq_table}`
"""
bq_max_ingest_ts_df = client.query(query).to_dataframe()
bq_max_ingest_epoch = str(int(bq_max_ingest_ts_df['ts'][0].timestamp()*1000000))
bq_max_ingest_ts = bq_max_ingest_ts_df['ts'][0]
print('Preparing to pull records with ingestion time >', bq_max_ingest_ts)
if not os.path.exists(bq_max_ingest_epoch):
os.makedirs(bq_max_ingest_epoch)
print('Saving output frames to', bq_max_ingest_epoch)
- Аннотируйте кадры, используя следующий код:
import json
import base64
import numpy as np
from IPython.display import Image, display, HTML, clear_output
im_width = stream.get(cv2.CAP_PROP_FRAME_WIDTH)
im_height = stream.get(cv2.CAP_PROP_FRAME_HEIGHT)
dashdelta = datetime.now()
framedata = {}
cntext = lambda x: {y['entity']['labelString']: y['count'] for y in x}
try:
while True:
try:
annotations_df = client.query(f'''
SELECT ingestion_time, annotation
FROM `{bq_table}`
WHERE ingestion_time > TIMESTAMP("{bq_max_ingest_ts}")
''').to_dataframe()
except ValueError as e:
continue
bq_max_ingest_ts = annotations_df['ingestion_time'].max()
for _, row in annotations_df.iterrows():
with frame_buffer_lock:
frame_ts = np.asarray(list(frame_buffer.keys()))
delta_ts = np.abs(frame_ts - (row['ingestion_time'].timestamp() * 1000))
delta_tx_idx = delta_ts.argmin()
closest_ts_delta = delta_ts[delta_tx_idx]
closest_ts = frame_ts[delta_tx_idx]
if closest_ts_delta > frame_buffer_error_milliseconds: continue
image = frame_buffer[closest_ts]
annotations = json.loads(row['annotation'])
for box in annotations['identifiedBoxes']:
image = cv2.rectangle(
image,
(
int(box['normalizedBoundingBox']['xmin']*im_width),
int(box['normalizedBoundingBox']['ymin']*im_height)
),
(
int((box['normalizedBoundingBox']['xmin'] + box['normalizedBoundingBox']['width'])*im_width),
int((box['normalizedBoundingBox']['ymin'] + box['normalizedBoundingBox']['height'])*im_height)
),
(255, 0, 0), 2
)
img_filename = f"{bq_max_ingest_epoch}/{row['ingestion_time'].timestamp() * 1000}.png"
cv2.imwrite(img_filename, image)
binimg = base64.b64encode(cv2.imencode('.jpg', image)[1]).decode()
curr_framedata = {
'path': img_filename,
'timestamp_error': closest_ts_delta,
'counts': {
**{
k['annotation']['displayName'] : cntext(k['counts'])
for k in annotations['stats']["activeZoneCounts"]
},
'full-frame': cntext(annotations['stats']["fullFrameCount"])
}
}
framedata[img_filename] = curr_framedata
if (datetime.now() - dashdelta).total_seconds() > dashboard_update_delay_seconds:
dashdelta = datetime.now()
clear_output()
display(HTML(f'''
<h1>Queue Monitoring Application</h1>
<p>Live Feed of the queue camera:</p>
<p><img alt="" src="{img_filename}" style="float: left;"/></a></p>
<table border="1" cellpadding="1" cellspacing="1" style="width: 500px;">
<caption>Current Model Outputs</caption>
<thead>
<tr><th scope="row">Metric</th><th scope="col">Value</th></tr>
</thead>
<tbody>
<tr><th scope="row">Serving Area People Count</th><td>{curr_framedata['counts']['serving-zone']['Person']}</td></tr>
<tr><th scope="row">Queueing Area People Count</th><td>{curr_framedata['counts']['queue-zone']['Person']}</td></tr>
<tr><th scope="row">Total Area People Count</th><td>{curr_framedata['counts']['full-frame']['Person']}</td></tr>
<tr><th scope="row">Timestamp Error</th><td>{curr_framedata['timestamp_error']}</td></tr>
</tbody>
</table>
<p> </p>
'''))
except KeyboardInterrupt:
print('Stopping Live Monitoring')
- Остановите задачу аннотации, используя кнопку «Стоп» в строке меню блокнота.
- Вы можете вернуться к отдельным кадрам, используя следующий код:
from IPython.html.widgets import Layout, interact, IntSlider
imgs = sorted(list(framedata.keys()))
def loadimg(frame):
display(framedata[imgs[frame]])
display(Image(open(framedata[imgs[frame]]['path'],'rb').read()))
interact(loadimg, frame=IntSlider(
description='Frame #:',
value=0,
min=0, max=len(imgs)-1, step=1,
layout=Layout(width='100%')))
11. Поздравления
Поздравляем, вы закончили лабораторную работу!
Очистить
Чтобы избежать списания средств с вашей учетной записи Google Cloud за ресурсы, используемые в этом руководстве, либо удалите проект, содержащий ресурсы, либо сохраните проект и удалите отдельные ресурсы.
Удалить проект
Удаление отдельных ресурсов
Ресурсы
https://cloud.google.com/vision-ai/docs/overview
https://cloud.google.com/vision-ai/docs/occupancy-count-tutorial