1. Objetivos
Descripción general
Este codelab se enfocará en crear una aplicación de Vertex AI Vision de extremo a extremo para supervisar el tamaño de la cola con videos de venta minorista. Usaremos las funciones incorporadas de análisis de ocupación del modelo especializado previamente entrenado para capturar los siguientes aspectos:
- Cuenta la cantidad de personas de pie en la cola.
- Contar la cantidad de personas a las que se atiende en el mostrador.
Qué aprenderás
- Cómo crear una aplicación en Vertex AI Vision y cómo implementarla
- Cómo configurar una transmisión de RTSP con un archivo de video y, luego, transferirla a Vertex AI Vision con vaictl desde un notebook de Jupyter.
- Cómo usar el modelo de análisis de ocupación y sus diferentes funciones
- Cómo buscar videos en tu almacenamiento en el almacén de contenido multimedia de Vertex AI Vision.
- Cómo conectar el resultado a BigQuery, escribir una consulta en SQL para extraer estadísticas del resultado de JSON del modelo y usar el resultado para etiquetar y anotar el video original
Costo:
El costo total de la ejecución de este lab en Google Cloud es de aproximadamente $2.
2. Antes de comenzar
Crea un proyecto y habilita las APIs:
- En la consola de Google Cloud, en la página del selector de proyectos, selecciona o crea un proyecto de Google Cloud. Nota: Si no planeas conservar los recursos creados durante este procedimiento, crea un proyecto en lugar de seleccionar uno existente. Cuando termines, puedes borrar el proyecto y quitar todos los recursos asociados con él. Ir al selector de proyectos
- Asegúrate de que la facturación esté habilitada para tu proyecto de Cloud. Obtén información sobre cómo verificar si la facturación está habilitada en un proyecto.
- Habilita Compute Engine, la API de Vertex, la API de Notebook y la API de Vision AI. Habilitar las API
Crea una cuenta de servicio:
- En la consola de Google Cloud, ve a la página Crear cuenta de servicio. Ir a Crear cuenta de servicio
- Elige tu proyecto.
- Escribe un nombre en el campo Nombre de cuenta de servicio. La consola de Google Cloud completa el campo ID de cuenta de servicio en función de este nombre. En el campo Descripción de la cuenta de servicio, ingresa una descripción. Por ejemplo, Service account for quickstart.
- Haz clic en Crear y continuar.
- Para proporcionar acceso a tu proyecto, otorga los siguientes roles a tu cuenta de servicio:
- Vision AI > Editor de Vision AI
- Compute Engine > Administrador de instancias de Compute (beta)
- BigQuery > Administrador de BigQuery .
En la lista Seleccionar un rol, elige un rol. Para obtener roles adicionales, haz clic en Agregar otro rol y agrega cada rol adicional.
- Haz clic en Continuar.
- Haz clic en Listo para terminar de crear la cuenta de servicio. No cierres la ventana del navegador. La usarás en la próxima tarea.
3. Configura un notebook de Jupyter
Antes de crear una aplicación en Análisis de ocupación, debe registrar una transmisión que la aplicación pueda usar más adelante.
En este instructivo, crearás una instancia de notebook de Jupyter que aloje un video y enviarás los datos de ese video en streaming desde el notebook. Usamos el notebook de Jupyter, ya que ofrece flexibilidad para ejecutar comandos de shell y códigos personalizados de procesamiento previo y posterior en un solo lugar, lo que es muy bueno para la experimentación rápida. Usaremos este {i>notebook<i} para:
- Ejecuta el servidor rtsp como un proceso en segundo plano
- Ejecutar el comando vaictl como proceso en segundo plano
- Ejecuta consultas y código de procesamiento para analizar el resultado de las estadísticas de ocupación
Crea un notebook de Jupyter
El primer paso para enviar un video desde una instancia de notebook de Jupyter es crear el notebook con la cuenta de servicio que creaste en el paso anterior.
- En la consola, ve a la página de Vertex AI. Ir a Vertex AI Workbench
- Haz clic en Notebooks administrados por el usuario.
- Haz clic en Nuevo notebook > Tensorflow Enterprise 2.6 (con LTS) > Sin GPU
- Ingresa el nombre del notebook de Jupyter. Para obtener más información, consulta Convención de nomenclatura de recursos.
- Haz clic en OPCIONES AVANZADAS.
- Desplázate hacia abajo hasta Secciones de permisos.
- Desmarca la opción Usar la cuenta de servicio predeterminada de Compute Engine.
- Agrega el correo electrónico de la cuenta de servicio que creaste en el paso anterior. Y haz clic en Crear.
- Una vez creada la instancia, haz clic en ABRIR JUPYTERLAB.
4. Configurar un notebook para transmitir video
Antes de crear una aplicación en Análisis de ocupación, debe registrar una transmisión que la aplicación pueda usar más adelante.
En este instructivo, usaremos nuestra instancia de notebook de Jupyter para alojar un video, y tú enviarás los datos de ese video en streaming desde la terminal del notebook.
Descargar la herramienta de línea de comandos vaictl
- En la instancia abierta de JupyterLab, abre un Notebook desde el selector.
- Descarga la herramienta de línea de comandos de Vertex AI Vision (vaictl), la herramienta de línea de comandos del servidor rtsp y la herramienta open-cv con el siguiente comando en la celda del notebook:
!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. Transferir un archivo de video para transmitirlo
Después de configurar tu entorno de notebook con las herramientas de línea de comandos requeridas, puedes copiar un archivo de video de muestra y, luego, usar vaictl para transmitir los datos de video a tu app de análisis de ocupación.
Registra una transmisión nueva
- Haz clic en la pestaña de transmisiones en el panel izquierdo de Vertex AI Vision.
- Haz clic en el botón Registrarse en la parte superior
. - En el nombre de las Novedades, ingresa "queue-stream".
- En región, elige la misma región que seleccionaste durante la creación del notebook en el paso anterior.
- Haz clic en Register.
Copia un video de muestra en la VM
- En tu notebook, copia un video de muestra con el siguiente comando wget.
!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4
Transmite videos desde una VM y transfiere datos a tu transmisión
- Para enviar este archivo de video local al flujo de entrada de la app, usa el siguiente comando en la celda de tu notebook. Debes realizar las siguientes sustituciones de variables:
- PROJECT_ID: Es el ID del proyecto de Google Cloud.
- LOCATION: el ID de ubicación. Por ejemplo, us-central1. Para obtener más información, consulta Ubicaciones de Cloud.
- LOCAL_FILE: Es el nombre de un archivo de video local. Por ejemplo,
seq25_h264.mp4.
PROJECT_ID='<Your Google Cloud project ID>'
LOCATION='<Your stream location>'
LOCAL_FILE='seq25_h264.mp4'
STREAM_NAME='queue-stream'
- Inicia un servidor rtsp-simple en el que transmitimos el archivo de video con el protocolo 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)
- Usa la herramienta de línea de comandos ffmpeg para repetir el video en la transmisión 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)
- Usar la herramienta de línea de comandos vaictl para transmitir el video desde el URI del servidor rtsp a nuestra transmisión “queue-stream” de Vertex AI Vision que creaste en el paso anterior.
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)
Es posible que pasen unos 100 segundos entre el inicio de la operación de transferencia vaictl y el video que aparece en el panel.
Una vez que la transferencia de transmisión esté disponible, podrás ver el feed de video en la pestaña Transmisiones del panel de Vertex AI Vision si seleccionas la transmisión de la fila.
Ve a la pestaña Transmisiones.
6. Crea una aplicación
El primer paso es crear una app que procese tus datos. Se puede considerar que una aplicación es como una canalización automatizada que conecta los siguientes elementos:
- Transferencia de datos: Se transfiere un feed de video a una transmisión.
- Análisis de datos: Se puede agregar un modelo de IA(visualización informática) después de la transferencia.
- Almacenamiento de datos: Las dos versiones del feed de video (la transmisión original y la que procesó el modelo de IA) se pueden almacenar en un almacén de contenido multimedia.
En la consola de Google Cloud, una app se representa como un gráfico.
Cómo crear una app vacía
Para poder completar el gráfico de la app, primero debes crear una app vacía.
Crea una app en la consola de Google Cloud.
- Ve a la consola de Google Cloud.
- Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ir a la pestaña Aplicaciones
- Haz clic en el botón Crear.
- Ingresa "queue-app" como nombre de la app y elige tu región.
- Haz clic en Crear.
Cómo agregar nodos de componentes de la app
Después de crear la aplicación vacía, puedes agregar los tres nodos al gráfico de la app:
- Nodo de transferencia: Es el recurso de transmisión que transfiere datos enviados desde un servidor de video rtsp que creaste en el notebook.
- Nodo de procesamiento: Es el modelo de análisis de ocupación que actúa sobre los datos transferidos.
- Nodo de almacenamiento: Es el almacén de contenido multimedia que almacena los videos procesados y que funciona como almacén de metadatos. Los almacenes de metadatos incluyen información analítica sobre datos de video transferidos y, también, información inferida por los modelos de IA.
Agrega nodos componentes a tu app en la consola.
- Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ir a la pestaña Aplicaciones
Esto te llevará a la visualización gráfica de la canalización de procesamiento.
Agrega un nodo de transferencia de datos
- Para agregar un nodo de flujo de entrada, selecciona la opción Streams en la sección Connectors del menú lateral.
- En la sección Fuente del menú Novedades, selecciona Agregar transmisiones.
- En el menú Agregar transmisiones, elige queue-stream.
- Para agregar el flujo al gráfico de la aplicación, haz clic en Agregar flujos.
Agrega un nodo de procesamiento de datos
- Para agregar el nodo del modelo de recuento de ocupación, selecciona la opción Análisis de ocupación en la sección Modelos especializados del menú lateral.
- Deja la selección predeterminada de Personas. Desmarca la opción Vehículos si ya está seleccionada.
- En la sección Opciones avanzadas, haz clic en Crear líneas/zonas activas
- Dibuja las zonas activas con la herramienta Polígono para contar las personas que se encuentran en esa zona. Etiqueta la zona según corresponda
- Haz clic en la flecha hacia atrás en la parte superior.
- Haz clic en la casilla de verificación para agregar parámetros de configuración del tiempo de permanencia para detectar la congestión.
Agrega un nodo de almacenamiento de datos
- Para agregar el nodo de destino de salida (almacenamiento), selecciona la opción VIsion AI Warehouse en la sección Conectores del menú lateral.
- Haz clic en el conector de Vertex AI Warehouse para abrir el menú y, luego, en Conectar almacén.
- En el menú Conectar almacén, selecciona Crear almacén nuevo. Asígnale el nombre queue-warehouse al almacén y deja la duración del TTL en 14 días.
- Haz clic en el botón Crear para agregar el almacén.
7. Conecta el resultado a la tabla de BigQuery
Cuando agregues un conector de BigQuery a tu app de Vertex AI Vision, todos los resultados de los modelos de apps conectadas se transferirán a la tabla de destino.
Puedes crear tu propia tabla de BigQuery y especificarla cuando agregues un conector de BigQuery a la app, o bien dejar que la plataforma de apps de Vertex AI Vision cree la tabla automáticamente.
Creación automática de tablas
Si permites que la plataforma de apps de Vertex AI Vision cree automáticamente la tabla, puedes especificar esta opción cuando agregues el nodo del conector de BigQuery.
Las siguientes condiciones de conjuntos de datos y tablas se aplican si deseas utilizar la creación automática de tablas:
- Conjunto de datos: El nombre del conjunto de datos que se creó automáticamente es visionai_dataset.
- Tabla: El nombre de la tabla que se creó automáticamente es visionai_dataset.APPLICATION_ID.
- Manejo de errores:
- Si existe una tabla con el mismo nombre en el mismo conjunto de datos, no se crea automáticamente.
- Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ir a la pestaña Aplicaciones
- Selecciona Ver aplicación junto al nombre de tu aplicación en la lista.
- En la página del compilador de aplicaciones, selecciona BigQuery en la sección Conectores.
- Deja vacío el campo Ruta de BigQuery.
- En store metadata from: Selecciona solo “ocupation Analytics” y desmarca las transmisiones.
El gráfico final de la app debería verse de la siguiente manera:
8. Implementar tu app para usarla
Una vez que compilaste tu app de extremo a extremo con todos los componentes necesarios, el último paso para usarla es implementarla.
- Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ir a la pestaña Aplicaciones
- Selecciona Ver app junto a la app queue-app de la lista.
- En la página de Studio, haz clic en el botón Implementar.
- En el siguiente diálogo de confirmación, haz clic en Implementar. La operación de implementación puede tardar varios minutos en completarse. Cuando finalice la implementación, aparecerán marcas de verificación verdes junto a los nodos.
9. Busca contenido de video en el almacén de almacenamiento
Después de transferir datos de video a tu app de procesamiento, puedes ver datos de video analizados y realizar búsquedas en función de la información de las estadísticas de ocupación.
- Abre la pestaña Almacén del panel de Vertex AI Vision. Ir a la pestaña Almacenes
- Busca el almacén del almacén en cola en la lista y haz clic en Ver recursos.
- En la sección Cantidad de personas, establece el valor Mín. en 1 y el valor Máx. en 5.
- Para filtrar los datos de video procesados y almacenados en el almacén de medios de Vertex AI Vision, haz clic en Buscar.
Una vista de los datos de video almacenados que coinciden con los criterios de búsqueda en la consola de Google Cloud.
10. Anota y analiza el resultado con la tabla de BigQuery
- En el notebook, inicializa las siguientes variables en la celda.
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'
- Ahora, capturaremos los fotogramas de la transmisión rtsp con el siguiente código:
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')
- Extrae la información de anotación y la marca de tiempo de los datos de la tabla de BigQuery y crea un directorio para almacenar las imágenes de marco capturadas:
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)
- Anota los marcos con el siguiente código:
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')
- Detén la tarea de anotación con el botón Detener en la barra de menú del notebook.
- Puedes volver a visitar marcos individuales con el siguiente código:
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. Felicitaciones
¡Felicitaciones! Terminaste el lab.
Realiza una limpieza
Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los recursos usados en este instructivo, borra el proyecto que contiene los recursos o conserva el proyecto y borra los recursos individuales.
Borra el proyecto
Borra los recursos individuales
Recursos
https://cloud.google.com/vision-ai/docs/overview
https://cloud.google.com/vision-ai/docs/occupancy-count-tutorial