1. Obiettivi
Panoramica
Questo codelab si concentrerà sulla creazione di un'applicazione Vertex AI Vision end-to-end per monitorare le dimensioni della coda utilizzando filmati video di vendita al dettaglio. Utilizzeremo le funzionalità integrate del modello specializzato preaddestrato Analisi dell'occupazione per acquisire quanto segue:
- Conta il numero di persone in coda.
- Conta il numero di persone servite al bancone.
Cosa imparerai a fare
- Come creare un'applicazione in Vertex AI Vision ed eseguirne il deployment
- Come configurare uno stream RTSP utilizzando un file video e importare lo stream in Vertex AI Vision utilizzando vaictl da un notebook Jupyter.
- Come utilizzare il modello Occupancy Analytics e le sue diverse funzionalità.
- Come cercare video in Media Warehouse di Vertex AI Vision.
- Come collegare l'output a BigQuery, scrivere una query SQL per estrarre insight dall'output JSON del modello e utilizzare l'output per etichettare e annotare il video originale.
Costo:
Il costo totale per eseguire questo lab su Google Cloud è di circa 2 $.
2. Prima di iniziare
Crea un progetto e abilita le API:
- Nella console Google Cloud, nella pagina di selezione del progetto, seleziona o crea un progetto Google Cloud. Nota: se non prevedi di conservare le risorse che crei in questa procedura, crea un progetto invece di selezionarne uno già esistente. Una volta completata questa procedura, puoi eliminare il progetto e tutte le relative risorse. Vai al selettore di progetti
- Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è abilitata per un progetto.
- Abilita le API Compute Engine, Vertex, Notebook e Vision AI. Abilita le API
Crea un service account:
- Nella console Google Cloud, vai alla pagina Crea service account. Vai a Crea service account
- Seleziona il progetto.
- Nel campo Nome service account, inserisci un nome. La console Google Cloud compila il campo ID account di servizio in base a questo nome. Nel campo Descrizione service account, inserisci una descrizione. Ad esempio, Service account for quickstart.
- Fai clic su Crea e continua.
- Per fornire l'accesso al tuo progetto, concedi i seguenti ruoli al tuo service account:
- Vision AI > Editor Vision AI
- Compute Engine > Compute Instance Admin (beta)
- BigQuery > BigQuery Admin.
Nell'elenco Seleziona un ruolo, seleziona un ruolo. Per i ruoli aggiuntivi, fai clic su Aggiungi un altro ruolo e aggiungi ogni ruolo successivo.
- Fai clic su Continua.
- Fai clic su Fine per completare la creazione del service account. Non chiudere la finestra del browser. Lo utilizzerai nel prossimo passaggio.
3. Configura Jupyter Notebook
Prima di creare un'app in Occupancy Analytics, devi registrare uno stream che potrà essere utilizzato in un secondo momento dall'app.
In questo tutorial creerai un'istanza di Jupyter Notebook che ospita un video e invierai i dati di streaming video dal notebook. Utilizziamo Jupyter Notebook perché ci offre la flessibilità di eseguire comandi shell e di eseguire codice di pre/post-elaborazione personalizzato in un unico posto, il che è molto utile per una sperimentazione rapida. Utilizzeremo questo notebook per:
- Esegui il server rtsp come processo in background
- Esegui il comando vaictl come processo in background
- Esegui query e codice di elaborazione per analizzare l'output dell'analisi dell'occupazione
Crea un blocco note Jupyter
Il primo passaggio per inviare video da un'istanza di Jupyter Notebook consiste nel creare il notebook con il service account creato nel passaggio precedente.
- Nella console, vai alla pagina Vertex AI. Vai a Vertex AI Workbench
- Fai clic su Notebooks gestiti dall'utente.
- Fai clic su Nuovo blocco note > Tensorflow Enterprise 2.6 (con LTS) > Senza GPU.
- Inserisci il nome del blocco note Jupyter. Per saperne di più, consulta le convenzioni per la denominazione delle risorse.
- Fai clic su OPZIONI AVANZATE.
- Scorri verso il basso fino alla sezione Sezioni delle autorizzazioni.
- Deseleziona l'opzione Usa il service account Compute Engine predefinito.
- Aggiungi l'email del service account creata nel passaggio precedente. e fai clic su Crea.
- Una volta creata l'istanza, fai clic su APRI JUPYTERLAB.
4. Configurare un notebook per lo streaming video
Prima di creare un'app in Occupancy Analytics, devi registrare uno stream che potrà essere utilizzato in un secondo momento dall'app.
In questo tutorial utilizzeremo la nostra istanza Jupyter Notebook per ospitare un video e invieremo i dati di streaming video dal terminale del notebook.
Scaricare lo strumento a riga di comando vaictl
- Nell'istanza di JupyterLab aperta, apri un notebook dal launcher.
- Scarica lo strumento a riga di comando Vertex AI Vision (vaictl), lo strumento a riga di comando del server rtsp e lo strumento open-cv utilizzando il seguente comando nella cella del blocco note:
!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. Importare un file video per lo streaming
Dopo aver configurato l'ambiente notebook con gli strumenti della riga di comando richiesti, puoi copiare un file video di esempio e poi utilizzare vaictl per trasmettere in streaming i dati video all'app di analisi dell'occupazione.
Registrare un nuovo stream
- Fai clic sulla scheda Flussi di clic nel riquadro a sinistra di Vertex AI Vision.
- Fai clic sul pulsante Registrati in alto
. - In Nome stream, inserisci "queue-stream".
- In Regione, scegli la stessa regione selezionata durante la creazione del notebook nel passaggio precedente.
- Fai clic su Registrati.
Copia un video di esempio nella VM
- Nel notebook, copia un video di esempio con il seguente comando wget.
!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4
Riprodurre in streaming video dalla VM e importare i dati nel flusso
- Per inviare questo file video locale al flusso di input dell'app, utilizza il seguente comando nella cella del notebook. Devi effettuare le seguenti sostituzioni di variabili:
- PROJECT_ID: l'ID del tuo progetto Google Cloud.
- LOCATION: il tuo ID località. Ad esempio, us-central1. Per maggiori informazioni, consulta Località cloud.
- LOCAL_FILE: il nome di un file video locale. Ad esempio,
seq25_h264.mp4.
PROJECT_ID='<Your Google Cloud project ID>'
LOCATION='<Your stream location>'
LOCAL_FILE='seq25_h264.mp4'
STREAM_NAME='queue-stream'
- Avvia un server rtsp-simple-server in cui trasmettiamo in streaming il file video con il protocollo 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)
- Utilizza lo strumento a riga di comando ffmpeg per riprodurre il video in loop nel flusso 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)
- Utilizza lo strumento a riga di comando vaictl per trasmettere in streaming il video dall'URI del server RTSP alla nostra coda di streaming "queue-stream" di Vertex AI Vision creata nel passaggio precedente.
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)
Potrebbero essere necessari circa 100 secondi tra l'avvio dell'operazione di importazione vaictl e la visualizzazione del video nella dashboard.
Una volta disponibile l'importazione dello stream, puoi visualizzare il feed video nella scheda Stream della dashboard di Vertex AI Vision selezionando lo stream della coda.
6. Crea un'applicazione
Il primo passaggio consiste nel creare un'app che elabori i tuoi dati. Un'app può essere considerata una pipeline automatizzata che collega:
- Importazione dei dati: un feed video viene importato in uno stream.
- Analisi dei dati: dopo l'importazione, è possibile aggiungere un modello di AI(Computer Vision).
- Archiviazione dei dati: le due versioni del feed video (lo stream originale e lo stream elaborato dal modello di AI) possono essere archiviate in un media warehouse.
Nella console Google Cloud, un'app è rappresentata come un grafico.
Creare un'app vuota
Prima di poter compilare il grafico dell'app, devi prima creare un'app vuota.
Crea un'app nella console Google Cloud.
- Vai alla console Google Cloud.
- Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
- Fai clic sul pulsante Crea.
- Inserisci "queue-app" come nome dell'app e scegli la tua regione.
- Fai clic su Crea.
Aggiungi nodi dei componenti dell'app
Dopo aver creato l'applicazione vuota, puoi aggiungere i tre nodi al grafico dell'app:
- Nodo di importione: la risorsa stream che importa i dati inviati da un server video RTSP creato nel blocco note.
- Nodo di elaborazione: il modello di analisi dell'occupazione che agisce sui dati importati.
- Nodo di archiviazione: il warehouse multimediale che archivia i video elaborati e funge da archivio dei metadati. I negozi di metadati includono informazioni di analisi sui dati video importati e informazioni dedotte dai modelli di AI.
Aggiungi nodi dei componenti all'app nella console.
- Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
Viene visualizzata la visualizzazione del grafico della pipeline di elaborazione.
Aggiungere un nodo di importazione dati
- Per aggiungere un nodo di flusso di input, seleziona l'opzione Flussi nella sezione Connettori del menu laterale.
- Nella sezione Origine del menu Stream che si apre, seleziona Aggiungi stream.
- Nel menu Aggiungi flussi, scegli queue-stream.
- Per aggiungere lo stream al grafico dell'app, fai clic su Aggiungi stream.
Aggiungere un nodo di elaborazione dei dati
- Per aggiungere il nodo del modello di conteggio dell'occupazione, seleziona l'opzione Analisi dell'occupazione nella sezione Modelli specializzati del menu laterale.
- Lascia le selezioni predefinite Persone. Deseleziona Veicoli se è già selezionata.
- Nella sezione Opzioni avanzate, fai clic su Crea zone/linee attive
. - Disegna le zone attive utilizzando lo strumento Poligono per conteggiare le persone in quella zona. Etichetta la zona di conseguenza
- Fai clic sulla freccia Indietro in alto.
- Aggiungi le impostazioni per il dwell time per rilevare la congestione facendo clic sulla casella di controllo.
Aggiungere un nodo di archiviazione dei dati
- Per aggiungere il nodo di destinazione di output (spazio di archiviazione), seleziona l'opzione Warehouse Vision AI nella sezione Connettori del menu laterale.
- Fai clic sul connettore Vertex AI Warehouse per aprire il menu, quindi fai clic su Connetti warehouse.
- Nel menu Connetti warehouse, seleziona Crea nuovo warehouse. Assegna al warehouse il nome queue-warehouse e lascia la durata TTL a 14 giorni.
- Fai clic sul pulsante Crea per aggiungere il warehouse.
7. Connetti l'output alla tabella BigQuery
Quando aggiungi un connettore BigQuery all'app Vertex AI Vision, tutti gli output del modello dell'app connessa vengono importati nella tabella di destinazione.
Puoi creare una tabella BigQuery e specificarla quando aggiungi un connettore BigQuery all'app oppure lasciare che la piattaforma dell'app Vertex AI Vision crei automaticamente la tabella.
Creazione automatica di tabelle
Se consenti alla piattaforma di app Vertex AI Vision di creare automaticamente la tabella, puoi specificare questa opzione quando aggiungi il nodo del connettore BigQuery.
Se vuoi utilizzare la creazione automatica delle tabelle, si applicano le seguenti condizioni per set di dati e tabelle:
- Set di dati: il nome del set di dati creato automaticamente è visionai_dataset.
- Tabella: il nome della tabella creato automaticamente è visionai_dataset.APPLICATION_ID.
- Gestione degli errori:
- Se esiste la tabella con lo stesso nome nello stesso set di dati, non viene creata automaticamente.
- Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
- Seleziona Visualizza app accanto al nome dell'applicazione nell'elenco.
- Nella pagina del generatore di applicazioni, seleziona BigQuery dalla sezione Connettori.
- Lascia vuoto il campo Percorso BigQuery.
- In Memorizza metadati da: seleziona solo "Analisi delle presenze" e deseleziona gli stream.
Il grafico finale dell'app dovrebbe essere simile a questo:
8. Esegui il deployment dell'app per utilizzarla
Dopo aver creato l'app end-to-end con tutti i componenti necessari, l'ultimo passaggio per utilizzarla è il deployment.
- Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
- Seleziona Visualizza app accanto all'app queue-app nell'elenco.
- Nella pagina Studio, fai clic sul pulsante Esegui il deployment.
- Nella finestra di dialogo di conferma successiva, fai clic su Esegui il deployment. Il completamento dell'operazione di deployment potrebbe richiedere diversi minuti. Al termine del deployment, accanto ai nodi vengono visualizzati segni di spunta verdi.
9. Cercare contenuti video nel warehouse di archiviazione
Dopo aver importato i dati video nell'app di elaborazione, puoi visualizzare i dati video analizzati e cercarli in base alle informazioni di analisi dell'occupazione.
- Apri la scheda Magazzini della dashboard di Vertex AI Vision. Vai alla scheda Magazzini
- Trova il magazzino di accodamento nell'elenco e fai clic su Visualizza asset.
- Nella sezione Conteggio persone, imposta il valore Min su 1 e il valore Max su 5.
- Per filtrare i dati video elaborati archiviati in Media Warehouse di Vertex AI Vision, fai clic su Cerca.
Una visualizzazione dei dati video archiviati che corrispondono ai criteri di ricerca nella console Google Cloud.
10. Annotare e analizzare l'output utilizzando la tabella BigQuery
- Nel notebook, inizializza le seguenti variabili nella cella.
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'
- Ora acquisiremo i frame dallo stream RTSP utilizzando il seguente codice:
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')
- Estrai il timestamp dei dati e le informazioni sulle annotazioni dalla tabella BigQuery e crea una directory per archiviare le immagini dei frame acquisiti:
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)
- Annota i frame utilizzando il seguente codice:
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')
- Interrompi l'attività di annotazione utilizzando il pulsante Interrompi nella barra dei menu del notebook
- Puoi rivedere i singoli frame utilizzando il seguente codice:
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. Complimenti
Complimenti, hai completato il lab.
Pulizia
Per evitare che al tuo account Google Cloud vengano addebitati costi relativi alle risorse utilizzate in questo tutorial, elimina il progetto che contiene le risorse oppure mantieni il progetto ed elimina le singole risorse.
Elimina il progetto
Elimina singole risorse
Risorse
https://cloud.google.com/vision-ai/docs/overview
https://cloud.google.com/vision-ai/docs/occupancy-count-tutorial