Application Vertex AI Vision Queue Detection

1. Objectifs

Présentation

Cet atelier de programmation porte sur la création de bout en bout d'une application Vertex AI Vision permettant de surveiller la taille de la file d'attente à l'aide de séquences vidéo commerciales. Nous utiliserons les caractéristiques intégrées du modèle spécialisé pré-entraîné Analyse de l'occupation pour capturer les éléments suivants:

• Comptez le nombre de personnes dans la file d'attente.
• Comptez le nombre de personnes servies au comptoir.

Points abordés

• Créer une application dans Vertex AI Vision et la déployer
• Configurer un flux RTSP à l'aide d'un fichier vidéo et l'ingérer dans Vertex AI Vision à l'aide de vaictl à partir d'un notebook Jupyter
• Utiliser le modèle d'analyse de l'occupation et ses différentes fonctionnalités
• Rechercher des vidéos dans l'entrepôt multimédia de Vertex AI Vision
• Comment connecter une sortie à BigQuery, écrire une requête SQL pour extraire des insights de la sortie JSON du modèle et l'utiliser pour étiqueter et annoter la vidéo d'origine.

Coût :

Le coût total d'exécution de cet atelier sur Google Cloud est d'environ 2 $.

2. Avant de commencer

Créez un projet et activez les API:

1. Dans la console Google Cloud, sur la page de sélection du projet, sélectionnez ou créez un projet Google Cloud. Remarque: Si vous ne prévoyez pas de conserver les ressources créées au cours de cette procédure, créez un projet au lieu de sélectionner un projet existant. Après avoir suivi ces étapes, vous pouvez supprimer le projet. Cela entraîne la suppression de toutes les ressources qui lui sont associées. Accéder au sélecteur de projet
2. Assurez-vous que la facturation est activée pour votre projet Cloud. Découvrez comment vérifier si la facturation est activée sur un projet.
3. Activez Compute Engine, l'API Vertex, l'API Notebook et l'API Vision AI. Activer les API

Créez un compte de service:

1. Dans la console Google Cloud, accédez à la page Créer un compte de service. Accéder à la page "Créer un compte de service"
2. Sélectionnez votre projet.
3. Dans le champ Nom du compte de service, saisissez un nom. La console Google Cloud renseigne le champ ID de compte de service à partir de ce nom. Dans le champ Description du compte de service, saisissez une description. Par exemple, "Compte de service" pour le guide de démarrage rapide.
4. Cliquez sur Créer et continuer.
5. Pour accorder l'accès à votre projet, attribuez le ou les rôles suivants à votre compte de service:

• Vision AI > Éditeur Vision AI
• Compute Engine > Administrateur d'instances Compute (bêta)
• BigQuery > Administrateur BigQuery .

Dans la liste Sélectionner un rôle, sélectionnez un rôle. Pour ajouter des rôles supplémentaires, cliquez sur Ajouter un autre rôle, puis ajoutez chaque rôle supplémentaire.

Remarque: Le champ "Rôle" détermine les ressources auxquelles votre compte de service peut accéder dans votre projet. Vous pouvez révoquer ces rôles ou attribuer des rôles supplémentaires ultérieurement. Dans les environnements de production, n'accordez pas les rôles Propriétaire, Éditeur ou Lecteur. Attribuez plutôt un rôle prédéfini ou personnalisé qui répond à vos besoins.

6. Cliquez sur Continuer.
7. Cliquez sur OK pour terminer la création du compte de service. Ne fermez pas la fenêtre de votre navigateur. Vous en aurez besoin lors de la tâche suivante.

3. Configurer un notebook Jupyter

Avant de créer une application dans l'analyse de l'occupation, vous devez enregistrer un flux que l'application pourra utiliser ultérieurement.

Dans ce tutoriel, vous allez créer une instance de notebook Jupyter qui héberge une vidéo, et vous allez envoyer ces données vidéo en flux continu depuis le notebook. Nous utilisons le notebook Jupyter, car il nous permet d'exécuter des commandes shell et d'exécuter du code personnalisé avant/après le traitement à un seul endroit, ce qui est très utile pour des tests rapides. Nous utiliserons ce notebook pour:

1. Exécuter le serveur rtsp en arrière-plan
2. Exécuter la commande vaictl en tant que processus en arrière-plan
3. Exécuter des requêtes et du code de traitement pour analyser les résultats de l'analyse de l'occupation

Créer un notebook Jupyter

Pour envoyer une vidéo à partir d'une instance de notebook Jupyter, la première étape consiste à créer le notebook avec le compte de service que vous avez créé à l'étape précédente.

1. Dans la console, accédez à la page "Vertex AI". Accéder à Vertex AI Workbench
2. Cliquer sur "Notebooks gérés par l'utilisateur"

3. Cliquez sur Nouveau notebook > TensorFlow Enterprise 2.6 (avec LTS) > Sans GPU

4. Saisissez le nom du notebook Jupyter. Pour en savoir plus, consultez Convention d'attribution de noms pour les ressources.

5. Cliquez sur OPTIONS AVANCÉES.
6. Faites défiler la page jusqu'à la section Autorisations.
7. Décochez l'option Utiliser le compte de service Compute Engine par défaut.
8. Ajoutez l'adresse e-mail du compte de service créé à l'étape précédente. Cliquez ensuite sur Créer.

9. Une fois l'instance créée, cliquez sur OUVRIR JUPYTERLAB.

4. Configurer un notebook pour diffuser des vidéos en streaming

Avant de créer une application dans l'analyse de l'occupation, vous devez enregistrer un flux que l'application pourra utiliser ultérieurement.

Dans ce tutoriel, nous allons utiliser notre instance de notebook Jupyter pour héberger une vidéo. Vous allez envoyer ces données vidéo en flux continu à partir du terminal Notebook.

Télécharger l'outil de ligne de commande vaictl

1. Dans l'instance Jupyterlab ouverte, ouvrez un notebook depuis le lanceur.

2. Téléchargez l'outil de ligne de commande Vertex AI Vision (vaictl), l'outil de ligne de commande du serveur rtsp ou l'outil open-cv à l'aide de la commande suivante dans la cellule du 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. Ingérer un fichier vidéo pour le streaming

Après avoir configuré votre environnement de notebook avec les outils de ligne de commande requis, vous pouvez copier un exemple de fichier vidéo, puis utiliser vaictl pour diffuser les données vidéo dans votre application d'analyse de l'occupation.

Enregistrer une nouvelle diffusion

1. Cliquez sur l'onglet "Flux" dans le panneau de gauche de Vertex AI Vision.
2. Cliquez sur le bouton "Enregistrer" en haut de la page.
3. Dans le nom du flux, saisissez 'queue-stream'.
4. Dans la région, choisissez la même région que celle sélectionnée lors de la création du notebook à l'étape précédente.
5. Cliquez sur Register (S'inscrire).

Buvez un café. L'inscription à la diffusion peut prendre jusqu'à une heure.

Copier un exemple de vidéo sur votre VM

1. Dans votre notebook, copiez un exemple de vidéo à l'aide de la commande wget suivante.

!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4

Diffuser des vidéos en streaming à partir d'une VM et ingérer des données dans votre flux

1. Pour envoyer ce fichier vidéo local au flux d'entrée de l'application, utilisez la commande suivante dans la cellule de votre notebook. Vous devez effectuer les substitutions de variables suivantes:

• PROJECT_ID: ID de votre projet Google Cloud.
• LOCATION: ID de zone géographique. Par exemple, us-central1. Pour en savoir plus, consultez la section Emplacements Cloud.
• LOCAL_FILE: nom d'un fichier vidéo local. Par exemple, seq25_h264.mp4.

PROJECT_ID='<Your Google Cloud project ID>'
LOCATION='<Your stream location>'
LOCAL_FILE='seq25_h264.mp4'
STREAM_NAME='queue-stream'

2. Démarrer un serveur rtsp-simple où nous diffuserons le fichier vidéo à l'aide du protocole 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. Utiliser l'outil de ligne de commande ffmpeg pour lire en boucle la vidéo dans un flux 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. Utiliser l'outil de ligne de commande vaictl pour diffuser la vidéo de l'URI du serveur rtsp vers notre flux "queue-stream" de Vertex AI Vision créé à l'étape précédente.

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)

Il peut s'écouler environ 100 secondes entre le démarrage de l'opération d'ingestion vaictl et l'affichage de la vidéo dans le tableau de bord.

Une fois l'ingestion du flux disponible, vous pouvez afficher le flux vidéo dans l'onglet Flux du tableau de bord Vertex AI Vision en sélectionnant le flux en file d'attente.

Accéder à l'onglet "Flux"

Crédit vidéo: ensemble de données Frames source, vidéo source

6. Créer une application

La première étape consiste à créer une application qui traite vos données. Une application peut être considérée comme un pipeline automatisé qui connecte les éléments suivants:

• Ingestion de données: un flux vidéo est ingéré dans un flux.
• Analyse des données: un modèle d'IA(Vision par ordinateur) peut être ajouté après l'ingestion.
• Stockage des données: les deux versions du flux vidéo (le flux d'origine et le flux traité par le modèle d'IA) peuvent être stockées dans un entrepôt multimédia.

Dans la console Google Cloud, une application est représentée sous forme de graphique.

Créer une application vide

Avant de pouvoir remplir le graphique de l'application, vous devez d'abord créer une application vide.

Créez une application dans la console Google Cloud.

1. Accédez à la console Google Cloud.
2. Ouvrez l'onglet Applications du tableau de bord Vertex AI Vision. Accéder à l'onglet "Applications"
3. Cliquez sur le bouton Créer.
4. Saisissez queue-app comme nom d'application et choisissez votre région.
5. Cliquez sur Créer.

Ajouter des nœuds de composants d'application

Une fois l'application vide créée, vous pouvez ajouter les trois nœuds au graphique de l'application:

1. Nœud d'ingestion: ressource de flux qui ingère les données envoyées depuis un serveur vidéo rtsp que vous avez créé dans le notebook.
2. Nœud de traitement: modèle d'analyse de l'occupation qui agit sur les données ingérées.
3. Nœud de stockage: entrepôt multimédia qui stocke les vidéos traitées et sert de magasin de métadonnées. Les magasins de métadonnées incluent des informations analytiques sur les données vidéo ingérées et des informations déduites par les modèles d'IA.

Ajoutez des nœuds de composants à votre application dans la console.

1. Ouvrez l'onglet Applications du tableau de bord Vertex AI Vision. Accéder à l'onglet "Applications"

La visualisation graphique du pipeline de traitement s'affiche.

Ajouter un nœud d'ingestion de données

1. Pour ajouter un nœud de flux d'entrée, sélectionnez l'option Flux dans la section Connecteurs du menu latéral.
2. Dans la section Source du menu Flux qui s'affiche, sélectionnez Ajouter des flux.
3. Dans le menu Ajouter des flux, sélectionnez queue-stream.
4. Pour ajouter le flux au graphique de l'application, cliquez sur Ajouter des flux.

Ajouter un nœud de traitement des données

1. Pour ajouter le nœud du modèle de nombre d'occupations, sélectionnez l'option d'analyse de l'occupation dans la section Modèles spécialisés du menu latéral.
2. Conservez les sélections par défaut Personnes. Décochez la case Véhicules si elle est déjà sélectionnée.

3. Dans la section "Options avancées", cliquez sur Créer des zones/lignes actives .
4. Tracez les zones actives à l'aide de l'outil Polygone pour compter les personnes présentes dans cette zone. Attribuez un libellé à la zone en conséquence.

5. Cliquez sur la flèche de retour en haut.

6. Ajoutez des paramètres de temps d’interaction afin de détecter les embouteillages en cochant la case correspondante.

Ajouter un nœud de stockage de données

1. Pour ajouter le nœud de destination de sortie (stockage), sélectionnez l'option VIsion AI Warehouse dans la section Connecteurs du menu latéral.
2. Cliquez sur le connecteur Vertex AI Warehouse pour ouvrir son menu, puis sur Connecter un entrepôt.
3. Dans le menu Connecter un entrepôt, sélectionnez Créer un entrepôt. Nommez l'entrepôt queue-warehouse et laissez la durée TTL à 14 jours.
4. Cliquez sur le bouton Créer pour ajouter l'entrepôt.

7. Connecter la sortie à la table BigQuery

Lorsque vous ajoutez un connecteur BigQuery à votre application Vertex AI Vision, toutes les sorties du modèle d'application connecté sont ingérées dans la table cible.

Vous pouvez créer votre propre table BigQuery et la spécifier lorsque vous ajoutez un connecteur BigQuery à l'application, ou laisser la plate-forme d'applications Vertex AI Vision créer automatiquement la table.

Création automatique de tableaux

Si vous autorisez la plate-forme d'applications Vertex AI Vision à créer automatiquement la table, vous pouvez spécifier cette option lorsque vous ajoutez le nœud du connecteur BigQuery.

Les conditions suivantes pour l'ensemble de données et la table s'appliquent si vous souhaitez utiliser la création automatique de tables:

• Ensemble de données: le nom de l'ensemble de données créé automatiquement est visionai_dataset.
• Table: le nom de la table créée automatiquement est visionai_dataset.APPLICATION_ID.
• Traitement des erreurs:
• Si une table portant le même nom sous le même ensemble de données existe, aucune création automatique n'est effectuée.

1. Ouvrez l'onglet Applications du tableau de bord Vertex AI Vision. Accéder à l'onglet "Applications"
2. Dans la liste, sélectionnez Afficher l'application à côté du nom de votre application.
3. Sur la page du générateur d'applications, sélectionnez BigQuery dans la section Connecteurs.
4. Laissez le champ Chemin d'accès BigQuery vide.

5. Dans les métadonnées du magasin de, sélectionnez uniquement Analyse de l'occupation et décochez les flux.

Le graphique final de l'application doit se présenter comme suit:

8. Déployer votre application pour l'utiliser

Une fois que vous avez créé votre application de bout en bout avec tous les composants nécessaires, la dernière étape pour l'utiliser consiste à la déployer.

1. Ouvrez l'onglet Applications du tableau de bord Vertex AI Vision. Accéder à l'onglet "Applications"
2. Sélectionnez View app (Afficher l'application) à côté de l'application queue-app dans la liste.
3. Sur la page Studio, cliquez sur le bouton Déployer.
4. Dans la boîte de dialogue de confirmation suivante, cliquez sur Déployer. L'opération de déploiement peut prendre plusieurs minutes. Une fois le déploiement terminé, des coches vertes apparaissent à côté des nœuds.

9. Rechercher du contenu vidéo dans l'entrepôt de stockage

Après avoir ingéré des données vidéo dans votre application de traitement, vous pouvez consulter les données vidéo analysées et effectuer des recherches sur la base des informations d'analyse de l'occupation.

1. Ouvrez l'onglet Entrepôts du tableau de bord Vertex AI Vision. Accéder à l'onglet "Entrepôts"
2. Recherchez l'entrepôt de l'entrepôt de file d'attente dans la liste, puis cliquez sur Afficher les éléments.
3. Dans la section Nombre de personnes, définissez la valeur Min sur 1 et la valeur Max sur 5.
4. Pour filtrer les données vidéo traitées stockées dans l'entrepôt multimédia de Vertex AI Vision, cliquez sur Rechercher.

Affichage des données vidéo stockées correspondant aux critères de recherche de la console Google Cloud.

10. Annoter et analyser le résultat à l'aide d'une table BigQuery

1. Dans le notebook, initialisez les variables suivantes dans la cellule.

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. Nous allons maintenant capturer les trames du flux rtsp à l'aide du code suivant:

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. Extrayez les informations d'annotation et d'horodatage des données de la table BigQuery, puis créez un répertoire pour stocker les images de frames capturées:

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. Annotez les frames à l'aide du code suivant:

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. Arrêtez la tâche d'annotation à l'aide du bouton Arrêter de la barre de menu du notebook.

6. Vous pouvez revoir des frames individuels à l'aide du code suivant:

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. Félicitations

Félicitations, vous avez terminé l'atelier !

Nettoyage

Pour éviter que les ressources utilisées lors de ce tutoriel soient facturées sur votre compte Google Cloud, supprimez le projet contenant les ressources, ou conservez le projet et supprimez chaque ressource individuellement.

Supprimer le projet

Supprimer des ressources individuelles

Ressources

https://cloud.google.com/vision-ai/docs/overview

https://cloud.google.com/vision-ai/docs/occupancy-count-tutorial

Licence

Enquête

Comment avez-vous utilisé ce tutoriel ?

Je vais seulement le lire Je l'ai lu et j'ai effectué les exercices

Cet atelier de programmation vous a-t-il été utile ?

Très utile Plutôt utile Pas utile

Cet atelier de programmation a-t-il été facile à suivre ?

Simple Modérée Difficile