Vertex AI Vision Queue 偵測應用程式

1. 目標

總覽

本程式碼研究室將著重於端對端建立 Vertex AI Vision 應用程式，使用零售影片片段監控佇列大小。我們會使用預先訓練的專業模型入座率分析內建功能來擷取下列資料：

  • 計算排隊人數。
  • 計算在櫃台服務的人數。

課程內容

  • 如何在 Vertex AI Vision 中建立及部署應用程式
  • 如何使用影片檔案設定 RTSP 串流，並透過 Jupyter Notebook 中的 vaictl 將串流取入 Vertex AI Vision。
  • 如何使用「Occupancy Analytics」模型及其各種功能。
  • 如何在 Vertex AI Vision 的媒體倉儲中搜尋影片。
  • 如何將輸出內容連結至 BigQuery、編寫 SQL 查詢，從模型的 JSON 輸出內容中擷取洞察資料，並使用輸出內容標記及標註原始影片。

費用：

在 Google Cloud 上執行本實驗室的總費用約為 $2 美元。

2. 事前準備

建立專案並啟用 API：

  1. 在 Google Cloud 控制台的專案選取器頁面中，選取或建立 Google Cloud 專案。注意事項：如果您不打算保留您在這項程序中建立的資源，請建立專案，而不要選取現有專案。這樣在完成這些步驟後，您就可以刪除專案，並移除與該專案相關聯的所有資源。前往專案選取器
  2. 確認 Cloud 專案已啟用計費功能。瞭解如何檢查專案是否已啟用計費功能。
  3. 啟用 Compute Engine、Vertex API、Notebooks API 和 Vision AI API。啟用 API

建立服務帳戶：

  1. 前往 Google Cloud 控制台的「Create service account」(建立服務帳戶) 頁面。前往「Create service account」(建立服務帳戶)
  2. 選取專案。
  3. 在「Service account name」(服務帳戶名稱) 欄位中輸入名稱。Google Cloud 控制台會根據這個名稱填入「Service account ID」欄位。在「服務帳戶說明」欄位中輸入說明。例如，快速入門指南的服務帳戶。
  4. 按一下 [建立並繼續]。
  5. 如要提供專案存取權，請將下列角色授予服務帳戶：
  • Vision AI > Vision AI 編輯器
  • Compute Engine > Compute Instance Admin (Beta 版)
  • BigQuery > BigQuery 管理員。

在「請選擇角色」清單中，選取角色。如需其他角色，請按一下「Add another role」(新增其他角色)，然後新增其他角色。

  1. 按一下「繼續」。
  2. 按一下「Done」，即完成建立服務帳戶。請勿關閉瀏覽器視窗。您會在下一個步驟中用到。

3. 設定 Jupyter Notebook

在 Occupancy Analytics 中建立應用程式前，您必須註冊可供應用程式日後使用的串流。

在本教學課程中，您會建立可代管影片的 Jupyter Notebook 執行個體，並透過筆記本傳送串流影片資料。我們使用 Jupyter Notebook，因為它可讓我們靈活執行 Shell 指令，並在單一位置執行自訂的前置/後置處理程式碼，非常適合快速實驗。我們會使用這個筆記本來：

  1. 以背景程序執行 rtsp 伺服器
  2. 以背景程序執行 vaictl 指令
  3. 執行查詢和處理程式碼，分析座位佔用率分析輸出

建立 Jupyter 筆記本

透過 Jupyter Notebook 執行階段傳送影片的第一步，是使用先前步驟建立的服務帳戶建立記事本。

  1. 前往控制台中的「Vertex AI」頁面。前往「Vertex AI Workbench」頁面
  2. 按一下「使用者自行管理的筆記本」

65b7112822858dce.png

  1. 依序點選「New Notebook」>「Tensorflow Enterprise 2.6 (with LTS)」>「Without GPUs」

dc156f20b14651d7.png

  1. 輸入 jupyter 筆記本名稱。詳情請參閱「資源命名慣例」。

b4dbc5fddc37e8d9.png

  1. 按一下「進階選項」
  2. 向下捲動至「權限」部分
  3. 取消勾選「Use Compute Engine default service account」(使用 Compute Engine 預設服務帳戶) 選項
  4. 新增先前步驟中建立的服務帳戶電子郵件地址。然後點選「建立」。

ec0b9ef00f0ef470.png

  1. 建立執行個體後，按一下「OPEN JUPYTERLAB」。

4. 設定 Notebook 以串流播放影片

在 Occupancy Analytics 中建立應用程式前，您必須註冊可供應用程式日後使用的串流。

在本教學課程中，我們會使用 Jupyter Notebook 執行個體來代管影片，而您則會透過 Notebook 終端機傳送串流影片資料。

下載 vaictl 指令列工具

  1. 在已開啟的 JupyterLab 例項中，透過啟動器開啟筆記本。

a6d182923ae4ada3.png

  1. 如要下載 Vertex AI Vision (vaictl) 指令列工具、rtsp 伺服器指令列工具、open-cv 工具，請在筆記本儲存格中使用下列指令：
!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. 擷取用於串流的影片檔案

使用必要的指令列工具設定筆記本環境後，可以複製範例影片檔案，然後使用 vaictl 將影片資料串流至入住人數分析應用程式。

註冊新的串流

  1. 按一下 Vertex AI Vision 左側面板中的「串流」分頁標籤。
  2. 按一下頂端的「Register」按鈕 eba418e723916514.png
  3. 在「Stream name」(串流名稱) 中輸入「queue-stream」
  4. 在「區域」中，選擇在先前步驟中建立 Notebook 時所選的區域。
  5. 按一下「註冊」

將範例影片複製到 VM

  1. 使用下列 wget 指令複製筆記本中的範例影片。
!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4

從 VM 串流播放影片，並將資料擷取至串流中

  1. 如要將這個本機影片檔案傳送至應用程式輸入串流，請在 Notebook 單元格中使用下列指令。您必須替換以下變數：
  • PROJECT_ID：您的 Google Cloud 專案 ID。
  • LOCATION：您的位置 ID。例如 us-central1詳情請參閱「Cloud 據點」。
  • LOCAL_FILE：本機影片檔案的檔案名稱。例如 seq25_h264.mp4。
PROJECT_ID='<Your Google Cloud project ID>'
LOCATION='<Your stream location>'
LOCAL_FILE='seq25_h264.mp4'
STREAM_NAME='queue-stream'
  1. 啟動 rtsp-simple-server，以便我們使用 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)
  1. 使用 ffmpeg 指令列工具，在 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)
  1. 使用 vaictl 指令列工具，將影片從 rtsp 伺服器 uri 串流至先前步驟中建立的 Vertex AI Vision 串流「queue-stream」。
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)

啟動 vaictl 攝入作業到影片顯示在資訊主頁之間，可能需要約 100 秒的時間。

串流攝入功能可供使用後，您可以選取佇列串流，在 Vertex AI Vision 資訊主頁的「串流」分頁中查看影片動態消息。

前往「訊息串」分頁

1b7aac7d36552f29.png

6. 建立應用程式

第一步是建立處理資料的應用程式。應用程式可視為連結以下項目的自動化管道：

  • 資料擷取：影片動態饋給會擷取至串流中。
  • 資料分析：在擷取後，您可以新增 AI(電腦視覺) 模型。
  • 資料儲存空間：兩個版本的影片動態饋給 (原始串流和 AI 模型處理的串流) 可儲存在媒體倉儲中。

在 Google Cloud 控制台中，應用程式會以圖表表示。

建立空白的應用程式

您必須先建立空白應用程式，才能填入應用程式圖表。

在 Google Cloud 控制台中建立應用程式。

  1. 前往 Google Cloud 控制台。
  2. 開啟 Vertex AI Vision 資訊主頁的「應用程式」分頁。前往「應用程式」分頁
  3. 按一下「建立」按鈕。21ecba7a23e9979e.png
  4. 輸入「queue-app」做為應用程式名稱，並選擇所屬區域。
  5. 按一下「建立」。

新增應用程式元件節點

建立空白應用程式後，您可以將三個節點新增至應用程式圖表：

  1. 擷取節點：擷取從 Notebook 中建立的 rtsp 影片伺服器傳送的資料的串流資源。
  2. 處理節點：處理已擷取資料的車輛乘載人數分析模型。
  3. 儲存節點：儲存經過處理的影片，並做為中繼資料儲存空間的媒體倉儲。中繼資料儲存項目包括已擷取影片資料的數據分析資訊，以及 AI 模型推測的資訊。

在控制台中為應用程式新增元件節點。

  1. 開啟 Vertex AI Vision 資訊主頁的「應用程式」分頁。前往「應用程式」分頁

這會帶您前往處理管道的圖表視覺化畫面。

新增資料擷取節點

  1. 如要新增輸入串流節點，請在側邊選單的「Connectors」(連接器) 區段中選取「Streams」(串流) 選項。
  2. 在隨即開啟的「Stream」選單的「Source」部分中，選取「Add Stream」。
  3. 在「Add streams」選單中，選擇「queue-stream」。
  4. 如要將串流新增至應用程式圖表，請按一下「新增串流」。

新增資料處理節點

  1. 如要新增入住人數模型節點，請在側邊選單的「Specialized models」(專用模型) 區段中，選取「occupancy analytics」(入住人數分析) 選項。
  2. 保留預設選項「使用者」。如果已勾選「車輛」，請取消勾選。

618b0c9dc671bae3.png

  1. 在「進階選項」部分，按一下「建立活動區域/線條」 5b2f31235603e05d.png
  2. 使用「多邊形」工具繪製活動區間，以計算該區域內的人。根據區域標示

50281a723650491f.png

  1. 按一下頂端的「返回」箭頭。

2bf0ff4d029d29eb.png

  1. 按一下「核取方塊」，新增停留時間設定，以便偵測壅塞情形。

c067fa256ca5bb96.png

新增資料儲存節點

  1. 如要新增輸出目的地 (儲存空間) 節點，請在側邊選單的「Connectors」(連接器) 區段中選取「Vision AI Warehouse」(Vision AI 倉儲) 選項。
  2. 按一下「Vertex AI Warehouse」連接器開啟選單，然後按一下「連結倉儲」。
  3. 在「Connect Warehouse」(連結倉儲) 選單中，選取「Create new warehouse」(新建倉儲)。將倉庫命名為「queue-warehouse」，並將存留時間保留為 14 天。
  4. 按一下「Create」(建立) 按鈕來新增倉儲。

7. 將輸出內容連結至 BigQuery 資料表

將 BigQuery 連接器新增至 Vertex AI Vision 應用程式後，所有已連結的應用程式模型輸出內容都會擷取至目標資料表。

您可以自行建立 BigQuery 資料表，並在將 BigQuery 連接器新增至應用程式時指定該資料表，或是讓 Vertex AI Vision 應用程式平台自動建立資料表。

自動建立資料表

如果您允許 Vertex AI Vision 應用程式平台自動建立資料表，您可以在新增 BigQuery 連接器節點時指定這個選項。

如要使用自動建立表格功能，請遵守下列資料集和表格條件：

  • 資料集：系統會自動建立名為 visionai_dataset 的資料集。
  • 資料表：系統自動建立的資料表名稱為 visionai_dataset.APPLICATION_ID。
  • 錯誤處理：
  • 如果在同一個資料集底下有同名的資料表，系統不會自動建立。
  1. 開啟 Vertex AI Vision 資訊主頁的「應用程式」分頁。前往「應用程式」分頁
  2. 從清單中選取應用程式名稱旁邊的「查看應用程式」。
  3. 在應用程式建構工具頁面中，從「連接器」專區選取「BigQuery」。
  4. 將「BigQuery 路徑」欄位留空。

ee0b67d4ab2263d.png

  1. 在「儲存中繼資料來源：」中，只選取「occupancy Analytics」，並取消選取串流。

最終的應用程式圖表應如下所示：

da0a1a049843572f.png

8. 部署應用程式以供使用

在使用所有必要元件建構端對端應用程式後，使用應用程式的最後一步就是部署應用程式。

  1. 開啟 Vertex AI Vision 資訊主頁的「應用程式」分頁。前往「應用程式」分頁
  2. 在清單中選取「queue-app」應用程式旁的「View app」。
  3. 在「Studio」頁面中，按一下「Deploy」按鈕。
  4. 在下列確認對話方塊中，按一下「部署」。部署作業可能需要幾分鐘才能完成。部署完成後，節點旁會顯示綠色勾號。dc514d9b9f35099d.png

9. 搜尋儲存倉庫中的影片內容

將影片資料擷取至處理應用程式後，您可以查看已分析的影片資料，並根據人流分析資訊搜尋資料。

  1. 開啟 Vertex AI Vision 資訊主頁的「倉庫」分頁。前往「倉庫」分頁
  2. 在清單中找出「queue-warehouse」倉庫，然後點選「View assets」。
  3. 在「人數」部分，將「最小」值設為 1，將「最大」值設為 5。
  4. 如要篩選儲存在 Vertex AI Vision 媒體倉儲中的已處理影片資料，請按一下「搜尋」。

a0e5766262443d6c.png

Google Cloud 主控台中儲存的影片資料符合搜尋條件的畫面。

10. 使用 BigQuery 資料表標記及分析輸出內容

  1. 在 Notebook 中，在儲存格中初始化下列變數。
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'
  1. 現在，我們要使用下列程式碼從 rtsp 串流中擷取影格：
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')
  1. 從 BigQuery 資料表提取資料時間戳記和註解資訊，並建立目錄來儲存擷取的畫面影像：
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)
  1. 使用下列程式碼為影格加上註解：
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')

9426ffe2376f0a7d.png

  1. 使用筆記本選單列中的「停止」按鈕停止註解工作

6c19cb00dcb28894.png

  1. 您可以使用下列程式碼查看個別影格：
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%')))

78b63b546a4c883b.png

11. 恭喜

恭喜，您已完成實驗室！

清理

如要避免系統向您的 Google Cloud 帳戶收取本教學課程所用資源的費用，請刪除含有相關資源的專案，或者保留專案但刪除個別資源。

刪除專案

刪除個別資源

資源

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

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

授權

問卷調查

您如何使用這個教學課程？

只能閱讀 閱讀並完成練習

您覺得這個程式碼研究室是否實用？

非常實用 還算實用 不實用

您覺得這個程式碼研究室有多容易理解？

簡單 中等 困難