Application Vertex AI Vision Queue Detection

1. Objectifs

Présentation

Cet atelier de programmation se concentre sur la création d'une application Vertex AI Vision de bout en bout pour surveiller la taille des files d'attente à l'aide de séquences vidéo de magasins. Nous allons utiliser les fonctionnalités 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 qui font la queue.
• Comptez le nombre de personnes servies au comptoir.

Points abordés

• Créer et déployer une application dans Vertex AI Vision
• Découvrez comment configurer un flux RTSP à l'aide d'un fichier vidéo et comment l'ingérer dans Vertex AI Vision à l'aide de vaictl à partir d'un notebook Jupyter.
• Découvrez comment utiliser le modèle Occupancy Analytics et ses différentes fonctionnalités.
• Découvrez comment rechercher des vidéos dans le Media Warehouse de Vertex AI Vision.
• Comment connecter la sortie à BigQuery, écrire une requête SQL pour extraire des insights à partir de la sortie JSON du modèle et utiliser la sortie 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 comptez pas conserver les ressources créées dans cette procédure, créez un projet au lieu d'en sélectionner un 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 les API Compute Engine, Vertex, Notebooks et 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 remplit le champ ID du compte de service en fonction de ce nom. Dans le champ Description du compte de service, saisissez une description. Par exemple, "Compte de service pour le 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 et 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'attribuez pas les rôles Propriétaire, Éditeur et 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 Jupyter Notebook

Avant de créer une application dans Occupancy Analytics, vous devez enregistrer un flux qui pourra être utilisé ultérieurement par l'application.

Dans ce tutoriel, vous allez créer une instance de notebook Jupyter qui héberge une vidéo, puis envoyer les données de streaming vidéo depuis le notebook. Nous utilisons un notebook Jupyter, car il nous offre la possibilité d'exécuter des commandes shell et d'exécuter du code de prétraitement/post-traitement personnalisé en un seul endroit, ce qui est très pratique pour une expérimentation rapide. Nous utiliserons ce notebook pour :

1. Exécuter le serveur rtsp en tant que processus d'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 des données analytiques sur l'occupation

Créer un notebook Jupyter

La première étape pour envoyer une vidéo à partir d'une instance Jupyter Notebook consiste à créer le notebook avec le compte de service que nous avons créé à l'étape précédente.

1. Dans la console, accédez à la page Vertex AI. Accéder à Vertex AI Workbench
2. Cliquez 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 Recommandation pour l'attribution de noms aux 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 Occupancy Analytics, vous devez enregistrer un flux qui pourra être utilisé ultérieurement par l'application.

Dans ce tutoriel, nous allons utiliser notre instance Jupyter Notebook pour héberger une vidéo et envoyer ces données vidéo en streaming depuis le 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 et l'outil OpenCV à 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

Une fois que vous avez 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 vers votre application d'analyse de l'occupation.

Enregistrer un nouveau flux

1. Cliquez sur l'onglet "Flux" dans le panneau de gauche de Vertex AI Vision.
2. Cliquez sur le bouton "S'inscrire" en haut .
3. Dans le champ "Nom du flux", saisissez queue-stream.
4. Dans "Région", sélectionnez la même région que celle choisie lors de la création du notebook à l'étape précédente.
5. Cliquez sur S'inscrire.

Prenez un café, l'enregistrement du flux 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 depuis 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 : votre ID d'établissement. Par exemple, us-central1. Pour en savoir plus, consultez 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émarrez un serveur rtsp-simple-server où nous diffusons le fichier vidéo avec le 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 la vidéo en boucle dans le 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. Utilisez l'outil de ligne de commande vaictl pour diffuser la vidéo depuis l'URI du serveur RTSP vers le flux Vertex AI Vision "queue-stream" 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ébut 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 voir le flux vidéo dans l'onglet Flux du tableau de bord Vertex AI Vision en sélectionnant le flux de file d'attente.

Accéder à l'onglet "Flux"

Crédits 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 relie 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 composant 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 inférées par les modèles d'IA.

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

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

Vous êtes alors redirigé vers la visualisation graphique du pipeline de traitement.

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 de modèle de comptage d'occupation, sélectionnez l'option Analyse de l'occupation dans la section Modèles spécialisés du menu latéral.
2. Conservez la sélection par défaut Personnes. Décochez Véhicules si cette option est déjà sélectionnée.

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

5. Cliquez sur la flèche de retour en haut de l'écran.

6. Ajoutez des paramètres pour le temps passé afin de détecter les embouteillages en cochant la case.

Ajouter un nœud de stockage de données

1. Pour ajouter le nœud de destination de sortie (stockage), sélectionnez l'option Entrepôt Vision AI dans la section Connecteurs du menu latéral.
2. Cliquez sur le connecteur Vertex AI Warehouse pour ouvrir son menu, puis cliquez 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 à une table BigQuery

Lorsque vous ajoutez un connecteur BigQuery à votre application Vertex AI Vision, toutes les sorties de modèle d'application connectées 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 de l'application Vertex AI Vision créer automatiquement la table.

Création automatique de tableaux

Si vous laissez la plate-forme d'application 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 s'appliquent aux ensembles de données et aux tables 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.
• Gestion des erreurs :
• Si une table portant le même nom existe déjà dans le même ensemble de données, aucune création automatique n'a lieu.

1. Ouvrez l'onglet Applications du tableau de bord Vertex AI Vision. Accéder à l'onglet "Applications"
2. Sélectionnez Afficher l'application à côté du nom de votre application dans la liste.
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 Stocker les métadonnées de, sélectionnez uniquement Analyse de l'occupation et décochez les flux.

Le graphique de l'application finale devrait 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 Afficher l'application à côté de l'application queue-app dans la liste.
3. Sur la page Studio, cliquez sur le bouton Deploy (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 s'affichent à côté des nœuds.

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

Une fois que vous avez ingéré des données vidéo dans votre application de traitement, vous pouvez afficher les données vidéo analysées et les rechercher en fonction 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 "queue-warehouse" dans la liste, puis cliquez sur Afficher les composants.
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 le Media Warehouse de Vertex AI Vision, cliquez sur Rechercher.

Vue des données vidéo stockées qui correspondent aux critères de recherche dans la console Google Cloud.

10. Annoter et analyser les résultats à 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 frames 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 l'horodatage des données et les informations d'annotation 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 dans la barre de menu du notebook.

6. Vous pouvez revenir sur 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 le lire uniquement Je vais le lire et effectuer les exercices

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

Très utile Moyennement utile Pas utile

Dans quelle mesure cet atelier de programmation était-il facile à suivre ?

Facile Moyenne Difficile