App de detección de colas de Vertex AI Vision

1. Objetivos

Descripción general

En este codelab, te enfocarás 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 integradas del modelo especializado Análisis de ocupación previamente entrenado para capturar lo siguiente:

• Cuenta la cantidad de personas de pie en la cola.
• Cuenta la cantidad de personas que reciben atención en el mostrador.

Qué aprenderás

• Cómo crear una aplicación en Vertex AI Vision y, luego, implementarla
• Cómo configurar una transmisión RTSP con un archivo de video y transferirla a Vertex AI Vision con vaictl desde un notebook de Jupyter
• Cómo usar el modelo de estadísticas de ocupación y sus diferentes funciones
• Cómo buscar videos en el almacén de contenido multimedia de tu almacenamiento 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 USD 2.

2. Antes de comenzar

Crea un proyecto y habilita las APIs:

1. En la página del selector de proyectos de la consola de Google Cloud, 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
2. 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.
3. Habilita las APIs de Compute Engine, Vertex, Notebook y Vision AI. Habilitar las API

Crea una cuenta de servicio:

1. En la consola de Google Cloud, ve a la página Crear cuenta de servicio. Ir a Crear cuenta de servicio
2. Elige tu proyecto.
3. 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.
4. Haz clic en Crear y continuar.
5. 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.

Nota: El campo Rol afecta a qué recursos puede acceder tu cuenta de servicio en tu proyecto. Puedes revocar estas funciones o asignar otras más adelante. En entornos de producción, no otorgues los roles de propietario, editor o visualizador. En su lugar, otorga un rol predefinido o personalizado que satisfaga tus necesidades.

6. Haz clic en Continuar.
7. 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 el notebook de Jupyter

Antes de crear una aplicación en Ocupancy Analytics, debes 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 Jupyter Notebook porque nos ofrece flexibilidad para ejecutar comandos de shell y ejecutar código de procesamiento previo o posterior personalizado en un solo lugar, lo que es muy bueno para la experimentación rápida. Usaremos este notebook para lo siguiente:

1. Ejecuta el servidor rtsp como un proceso en segundo plano
2. Ejecuta el comando vaictl como proceso en segundo plano
3. Ejecuta consultas y procesa código para analizar el resultado de las estadísticas de ocupación

Cómo crear 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.

1. En la consola, ve a la página de Vertex AI. Ir a Vertex AI Workbench
2. Haz clic en Notebooks administrados por el usuario.

3. Haz clic en Nuevo notebook > Tensorflow Enterprise 2.6 (with LTS) > Without GPUs.

4. Ingresa el nombre del notebook de Jupyter. Para obtener más información, consulta Convención de nombres de recursos.

5. Haz clic en OPCIONES AVANZADAS.
6. Desplázate hacia abajo hasta Secciones de permisos.
7. Desmarca la opción Usar la cuenta de servicio predeterminada de Compute Engine.
8. Agrega el correo electrónico de la cuenta de servicio que creaste en el paso anterior. Y haz clic en Crear.

9. Una vez que se cree la instancia, haz clic en OPEN JUPYTERLAB.

4. Cómo configurar un notebook para transmitir videos

Antes de crear una aplicación en Ocupancy Analytics, debes 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.

Descarga la herramienta de línea de comandos vaictl

1. En la instancia abierta de JupyterLab, abre un notebook desde el selector.

2. 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. Transfiere un archivo de video para transmitirlo

Después de configurar el entorno de tu notebook con las herramientas de línea de comandos necesarias, puedes copiar un archivo de video de muestra y, luego, usar vaictl para transmitir los datos de video a tu app de estadísticas de ocupación.

Registra una transmisión nueva

1. Haz clic en la pestaña Flujos en el panel izquierdo de Vertex AI Vision.
2. Haz clic en el botón Registrarse en la parte superior .
3. En el nombre de la transmisión, ingresa "queue-stream".
4. En Región, elige la misma región que seleccionaste durante la creación del notebook en el paso anterior.
5. Haz clic en Registrar.

Tómate un café, ya que la transmisión puede tardar hasta 1 hora en registrarse.

Copia un video de muestra en la VM

1. 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 la VM y transfiere datos a tu transmisión

1. 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: El ID de tu proyecto de Google Cloud.
• LOCATION: Es el ID de tu 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'

2. Inicia un rtsp-simple-server 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)

3. Usa la herramienta de línea de comandos ffmpeg para reproducir en bucle el video en la transmisión de 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)

4. Usa la herramienta de línea de comandos vaictl para transmitir el video desde el URI del servidor RTSP a la transmisión de Vertex AI Vision "queue-stream" 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 demore alrededor de 100 segundos entre el inicio de la operación de transferencia de vaictl y la aparición del video en el panel.

Una vez que la transferencia de transmisiones esté disponible, podrás ver el feed de video en la pestaña Streams del panel de Vertex AI Vision seleccionando el flujo de transmisión de fila.

Ve a la pestaña Flujos.

Crédito del video: fuente del conjunto de datos de fotogramas, fuente de video

6. Crea una aplicación

El primer paso es crear una app que procese tus datos. Una app se puede considerar como una canalización automatizada que conecta lo siguiente:

• 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

Antes de propagar el gráfico de la app, primero debes crear una app vacía.

Crea una app en la consola de Google Cloud.

1. Ve a la consola de Google Cloud.
2. Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ve a la pestaña Aplicaciones.
3. Haz clic en el botón Crear.
4. Ingresa "queue-app" como el nombre de la app y elige tu región.
5. Haz clic en Crear.

Agrega 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:

1. Nodo de transferencia: Es el recurso de transmisión que transfiere datos enviados desde un servidor de video rtsp que creaste en el notebook.
2. Nodo de procesamiento: Es el modelo de análisis de ocupación que actúa sobre los datos transferidos.
3. Nodo de almacenamiento: Es el almacén de contenido multimedia que almacena los videos procesados y funciona como un almacén de metadatos. Los almacenes de metadatos incluyen información de análisis sobre los datos de video transferidos y la información inferida por los modelos de IA.

Agrega nodos componentes a tu app en la consola.

1. Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ir a la pestaña Aplicaciones

Esto te llevará a la visualización del gráfico de la canalización de procesamiento.

Cómo agregar un nodo de transferencia de datos

1. Para agregar un nodo de flujo de entrada, selecciona la opción Flujos en la sección Conectores del menú lateral.
2. En la sección Fuente del menú Transmisiones que se abre, selecciona Agregar transmisiones.
3. En el menú Agregar transmisiones, elige queue-stream.
4. Para agregar el flujo al gráfico de la aplicación, haz clic en Agregar transmisiones.

Agrega un nodo de procesamiento de datos

1. 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.
2. Deja la selección predeterminada de Personas. Desmarca Vehículos si ya está seleccionado.

3. En la sección Opciones avanzadas, haz clic en Crear zonas o líneas activas
4. Dibuja las zonas activas con la herramienta Polígono para registrar a las personas que se encuentran en esa zona. Etiqueta la zona según corresponda

5. Haz clic en la flecha hacia atrás en la parte superior.

6. Para agregar parámetros de configuración del tiempo de permanencia y detectar la congestión, haz clic en la casilla de verificación.

Cómo agregar un nodo de almacenamiento de datos

1. Para agregar el nodo de destino de salida (almacenamiento), selecciona la opción almacén de Vision AI en la sección Conectores del menú lateral.
2. Haz clic en el conector Vertex AI Warehouse para abrir su menú y, luego, en Connect warehouse.
3. En el menú Conectar almacén, selecciona Crear almacén nuevo. Asigna el nombre queue-warehouse al almacén y deja la duración del TTL en 14 días.
4. Haz clic en el botón Crear para agregar el almacén.

7. Cómo conectar la salida a la tabla de BigQuery

Cuando agregues un conector de BigQuery a tu app de Vertex AI Vision, todos los resultados del modelo de app conectado 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 permitir 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 la tabla automáticamente, puedes especificar esta opción cuando agregues el nodo del conector de BigQuery.

Se aplican las siguientes condiciones del conjunto de datos y la tabla si deseas usar la creación automática de tablas:

• Dataset: El nombre del conjunto de datos creado automáticamente es visionai_dataset.
• Tabla: El nombre de la tabla creada 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 realizará la creación automática.

1. Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ve a la pestaña Aplicaciones.
2. Selecciona Ver app junto al nombre de tu aplicación en la lista.
3. En la página del compilador de aplicaciones, selecciona BigQuery en la sección Conectores.
4. Deja vacío el campo Ruta de BigQuery.

5. En Almacenar metadatos de:, selecciona solo "Analytics de ocupación" 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.

1. Abre la pestaña Aplicaciones del panel de Vertex AI Vision. Ve a la pestaña Aplicaciones.
2. Selecciona Ver app junto a la app queue-app en la lista.
3. En la página Studio, haz clic en el botón Implementar.
4. 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 finaliza la implementación, aparecen 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 los datos de video analizados y buscarlos en función de la información de estadísticas de ocupación.

1. Abre la pestaña Almacenes del panel de Vision de Vertex AI. Ve a la pestaña Almacenes
2. Busca el almacén queue-warehouse en la lista y haz clic en Ver recursos.
3. En la sección Cantidad de personas, establece el valor de Mín. en 1 y el de Máx. en 5.
4. Para filtrar los datos de video procesados almacenados en Media Warehouse de Vertex AI Vision, haz clic en Búsqueda.

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

1. 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'

2. 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')

3. Extrae la marca de tiempo de los datos y la información de anotación de la tabla de BigQuery y crea un directorio para almacenar las imágenes de fotogramas 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)

4. 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>&nbsp;</p>
        '''))
except KeyboardInterrupt:
  print('Stopping Live Monitoring')

5. Detén la tarea de anotación con el botón Detener en la barra de menú del notebook.

6. 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! Completaste 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

Licencia

Encuesta

¿Cómo usaste este instructivo?

Leer Leer y completar los ejercicios

¿Qué tan útil te resultó este codelab?

Muy útil Moderadamente útil No útil

¿Qué tan fácil fue seguir este codelab?

Fácil Moderado Difícil