App Vertex AI Vision per il rilevamento delle code

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 le seguenti informazioni:

• Conta il numero di persone in coda.
• Conta il numero di persone servite al banco.

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 importarlo in Vertex AI Vision utilizzando vaictl da un Jupyter Notebook.
• Come utilizzare il modello di Dati sull'occupazione e le sue diverse funzionalità.
• Come cercare video nello spazio di archiviazione Media Warehouse di Vertex AI Vision.
• Come collegare l'output a BigQuery, scrivere query SQL per estrarre informazioni 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:

1. 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 create in questa procedura, crea un progetto invece di selezionarne uno esistente. Una volta completata questa procedura, puoi eliminare il progetto e tutte le relative risorse. Vai al selettore dei progetti
2. Verifica che la fatturazione sia attivata per il tuo progetto Cloud. Scopri come verificare se la fatturazione è attivata per un progetto.
3. Abilita le API Compute Engine, Vertex, Notebook e Vision AI. Abilita le API

Crea un account di servizio:

1. Nella console Google Cloud, vai alla pagina Crea account di servizio. Vai a Crea account di servizio
2. Seleziona il progetto.
3. Inserisci un nome nel campo Nome account di servizio. La console Google Cloud compila il campo ID account di servizio in base a questo nome. Nel campo Descrizione account di servizio, inserisci una descrizione. ad esempio Account di servizio per la guida rapida.
4. Fai clic su Crea e continua.
5. Per fornire l'accesso al progetto, concedi i seguenti ruoli al tuo account di servizio:

• Vision AI > Editor Vision AI
• Compute Engine > Amministratore istanze Compute (beta)
• BigQuery > BigQuery Admin.

Nell'elenco Seleziona un ruolo, seleziona un ruolo. Per altri ruoli, fai clic su Aggiungi un altro ruolo e aggiungi ogni ruolo aggiuntivo.

Nota: il campo Ruolo influisce sulle risorse a cui può accedere il tuo account di servizio nel progetto. Puoi revocare questi ruoli o concederne altri in un secondo momento. Negli ambienti di produzione, non concedere i ruoli di Proprietario, Editor o Visualizzatore. Concedi invece un ruolo predefinito o personalizzato che soddisfi le tue esigenze.

6. Fai clic su Continua.
7. Fai clic su Fine per completare la creazione dell'account di servizio. Non chiudere la finestra del browser. Lo utilizzerai nel passaggio successivo.

3. Configurare Jupyter Notebook

Prima di creare un'app in Dati sull'occupazione, devi registrare uno stream che potrà essere utilizzato in un secondo momento dall'app.

In questo tutorial creerai un'istanza di blocco note Jupyter che ospita un video e invierai i dati video in streaming dal blocco note. Utilizziamo Jupyter Notebook perché ci offre la flessibilità di eseguire comandi shell ed eseguire codice di pre/post-elaborazione personalizzato in un unico posto, il che è molto utile per la sperimentazione rapida. Utilizzeremo questo blocco note per:

1. Esegui il server rtsp come processo in background
2. Esegui il comando vaictl come processo in background
3. Esegui query ed elabora il codice per analizzare l'output dell'analisi dell'occupazione

Creare un blocco note Jupyter

Il primo passaggio per inviare un video da un'istanza del Jupyter Notebook consiste nel creare il notebook con il nostro account di servizio creato nel passaggio precedente.

1. Nella console, vai alla pagina di Vertex AI. Vai a Vertex AI Workbench
2. Fai clic su Blocchi note gestiti dall'utente.

3. Fai clic su Nuovo blocco note > Tensorflow Enterprise 2.6 (con LTS) > Senza GPU.

4. Inserisci il nome per il blocco note Jupyter. Per ulteriori informazioni, consulta la convenzione di denominazione delle risorse.

5. Fai clic su OPZIONI AVANZATE.
6. Scorri verso il basso fino a Sezioni delle autorizzazioni.
7. Deseleziona l'opzione Utilizza l'account di servizio predefinito di Compute Engine.
8. Aggiungi l'indirizzo email dell'account di servizio creato nel passaggio precedente. Infine, fai clic su Crea.

9. Dopo aver creato l'istanza, fai clic su APRI JUPYTERLAB.

4. Configura un blocco note per riprodurre video in streaming

Prima di creare un'app in Dati sull'occupazione, devi registrare uno stream che potrà essere utilizzato in un secondo momento dall'app.

In questo tutorial utilizzeremo la nostra istanza di blocco note Jupyter per ospitare un video e tu invierai i dati video in streaming dal terminale Notebook.

Scarica lo strumento a riga di comando vaictl

1. Nell'istanza di JupyterLab aperta, apri un notebook dal programma di avvio.

2. Scarica lo strumento a riga di comando Vertex AI Vision (vaictl), lo strumento a riga di comando rtsp server 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 del notebook con gli strumenti a riga di comando richiesti, puoi copiare un file video di esempio e utilizzare vaictl per trasmettere in streaming i dati video alla tua app di analisi della presenza.

Registrare un nuovo stream

1. Fai clic sulla scheda Stream nel riquadro a sinistra di Vertex AI Vision.
2. Fai clic sul pulsante Registrati in alto
3. In Nome stream, inserisci "queue-stream"
4. In Regione, scegli la stessa regione selezionata durante la creazione del blocco note nel passaggio precedente.
5. Fai clic su Registrati.

Prenditi un caffè, la registrazione dello stream può richiedere fino a un'ora.

Copiare un video di esempio nella VM

1. 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 nello stream

1. Per inviare questo file video locale allo stream di input dell'app, utilizza il seguente comando nella cella del notebook. Devi apportare le seguenti sostituzioni delle variabili:

• PROJECT_ID: l'ID del tuo progetto Google Cloud.
• LOCATION: l'ID della tua località. Ad esempio, us-central1. Per ulteriori 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'

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

3. Utilizza lo strumento a riga di comando ffmpeg per riprodurre in loop il video nello stream 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. Utilizza lo strumento a riga di comando vaictl per riprodurre in streaming il video dall'URI del server RTSP allo stream "queue-stream" di Vertex AI Vision creato 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.

Quando l'importazione dello stream è disponibile, puoi visualizzare il feed video nella scheda Stream della dashboard di Vertex AI Vision selezionando lo stream in coda.

Vai alla scheda Stream

Riconoscimenti del video: origine del set di dati Frame, origine del video

6. Crea un'applicazione

Il primo passaggio consiste nel creare un'app che elabori i dati. Un'app può essere considerata una pipeline automatizzata che collega quanto segue:

• Importazione dei dati: un feed video viene importato in uno stream.
• Analisi dei dati: un modello di IA(Visione artificiale) può essere aggiunto dopo l'importazione.
• Archiviazione dei dati: le due versioni del feed video (lo stream originale e lo stream elaborato dal modello di IA) 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 grafo delle app, devi creare un'app vuota.

Crea un'app nella console Google Cloud.

1. Vai alla console Google Cloud.
2. Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
3. Fai clic sul pulsante Crea.
4. Inserisci "queue-app" come nome dell'app e scegli la tua regione.
5. Fai clic su Crea.

Aggiungere nodi dei componenti dell'app

Dopo aver creato l'applicazione vuota, puoi aggiungere i tre nodi al grafo dell'app:

1. Nodo di importazione: la risorsa stream che importa i dati inviati da un server video RTSP creato nel notebook.
2. Nodo di elaborazione: il modello di analisi delle presenze che agisce sui dati importati.
3. Nodo di archiviazione: il warehouse multimediale che archivia i video elaborati e funge da archivio dei metadati. I datastore dei metadati includono informazioni di analisi sui dati video importati e informazioni dedotte dai modelli di IA.

Aggiungi nodi dei componenti all'app nella console.

1. Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni

Viene visualizzata la visualizzazione grafica della pipeline di elaborazione.

Aggiungere un nodo di importazione dati

1. Per aggiungere un nodo di stream di input, seleziona l'opzione Stream nella sezione Connettori del menu laterale.
2. Nella sezione Origine del menu Stream che si apre, seleziona Aggiungi flussi.
3. Nel menu Aggiungi flussi, scegli queue-stream.
4. Per aggiungere lo stream al grafico dell'app, fai clic su Aggiungi flussi.

Aggiungere un nodo di elaborazione dei dati

1. Per aggiungere il nodo del modello di conteggio del numero di persone, seleziona l'opzione Analisi del numero di persone nella sezione Modelli specializzati del menu laterale.
2. Lascia le selezioni predefinite Persone. Deseleziona Veicoli se è già selezionata.

3. Nella sezione Opzioni avanzate, fai clic su Crea zone/linee attive .
4. Disegna le zone attive utilizzando lo strumento Poligono per conteggiare le persone all'interno di quella zona. Etichetta la zona di conseguenza

5. Fai clic sulla Freccia Indietro in alto.

6. Aggiungi le impostazioni per il tempo di attesa per rilevare la congestione facendo clic su Casella di controllo.

Aggiungere un nodo di archiviazione dati

1. Per aggiungere il nodo di destinazione di output (spazio di archiviazione), seleziona l'opzione Warehouse Vision AI nella sezione Connettori del menu laterale.
2. Fai clic sul connettore Vertex AI Warehouse per aprire il relativo menu, poi fai clic su Connetti warehouse.
3. Nel menu Connetti warehouse, seleziona Crea nuovo warehouse. Assegna al magazzino il nome queue-warehouse e lascia la durata TTL a 14 giorni.
4. Fai clic sul pulsante Crea per aggiungere il warehouse.

7. Collega l'output alla tabella BigQuery

Quando aggiungi un connettore BigQuery all'app Vertex AI Vision, tutti gli output del modello dell'app collegati verranno importati nella tabella di destinazione.

Puoi creare una tabella BigQuery e specificarla quando aggiungi un connettore BigQuery all'app oppure lasciare che sia la piattaforma dell'app Vertex AI Vision a creare 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 i set di dati e le tabelle:

• Set di dati: il nome del set di dati creato automaticamente è visionai_dataset.
• Tabella: il nome della tabella creata automaticamente è visionai_dataset.APPLICATION_ID.
• Gestione degli errori:
• Se esiste una tabella con lo stesso nome sotto lo stesso set di dati, non viene creata alcuna creazione automatica.

1. Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
2. Seleziona Visualizza app accanto al nome dell'applicazione nell'elenco.
3. Nella pagina di creazione dell'applicazione, seleziona BigQuery dalla sezione Connettori.
4. Lascia vuoto il campo Percorso BigQuery.

5. In Memorizza i metadati in: seleziona solo "Analytics sulla presenza" e deseleziona gli stream.

Il grafico dell'app finale dovrebbe avere il seguente aspetto:

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.

1. Apri la scheda Applicazioni della dashboard di Vertex AI Vision. Vai alla scheda Applicazioni
2. Seleziona Visualizza app accanto all'app queue-app nell'elenco.
3. Nella pagina Studio, fai clic sul pulsante Esegui il deployment.
4. Nella seguente finestra di dialogo di conferma, 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 dei segni di spunta verdi.

9. Cercare i contenuti video nel magazzino di archiviazione

Dopo aver importato i dati video nella tua app di elaborazione, puoi visualizzare i dati video analizzati ed eseguire ricerche nei dati in base alle informazioni sull'analisi del numero di persone.

1. Apri la scheda Magazzini della dashboard di Vertex AI Vision. Vai alla scheda Magazzini
2. Individua il warehouse code-warehouse nell'elenco e fai clic su Visualizza asset.
3. Nella sezione Numero di persone, imposta il valore Min su 1 e il valore Max su 5.
4. Per filtrare i dati video elaborati archiviati nel 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

1. Nel blocco note, 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'

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

3. Estrai il timestamp dei dati e le informazioni di annotazione dalla tabella BigQuery e crea una directory in cui archiviare le immagini del frame acquisite:

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

5. Interrompi l'attività di annotazione utilizzando il pulsante Arresta nella barra dei menu del blocco note

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

Congratulazioni, hai terminato 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

Licenza

Sondaggio

Come hai utilizzato questo tutorial?

Solo lettura Leggi e completa gli esercizi

Quanto è stato utile questo codelab?

Molto utile Moderatamente utile Non utile

Quanto è stato facile seguire questo codelab?

Facile Moderata Difficile