1. Cele
Omówienie
To ćwiczenie w programowaniu skupia się na kompleksowym tworzeniu aplikacji Vertex AI Vision pod kątem monitorowania rozmiaru kolejki za pomocą nagrań wideo ze sprzedaży detalicznej. Użyjemy już wytrenowanego modelu specjalistycznego analizy zajętości, aby rejestrować te dane:
- Policz liczbę osób stojących w kolejce.
- Policz, ile osób jest obsługiwanych przy ladzie.
Czego się nauczysz
- Jak utworzyć aplikację w Vertex AI Vision i wdrożyć ją
- Jak skonfigurować strumień RTSP za pomocą pliku wideo i pozyskać go w Vertex AI Vision za pomocą vaictl z notatnika Jupyter.
- Jak korzystać z modelu analizy zajętości i jego różnych funkcji.
- Jak wyszukiwać filmy w hurtowni multimediów Vertex AI Vision.
- Jak połączyć dane wyjściowe z BigQuery, napisz zapytanie SQL w celu wyodrębnienia obserwacji z danych wyjściowych modelu JSON oraz użyj tych danych do oznaczenia oryginalnego filmu i adnotacji.
Koszt:
Całkowity koszt uruchomienia tego modułu w Google Cloud wynosi około 2 USD.
2. Zanim zaczniesz
Utwórz projekt i włącz interfejsy API:
- W konsoli Google Cloud na stronie selektora projektów wybierz lub utwórz projekt Google Cloud. Uwaga: jeśli nie planujesz zachować zasobów utworzonych w ramach tej procedury, zamiast wybierać istniejący projekt, utwórz projekt. Po wykonaniu tych czynności możesz usunąć projekt wraz ze wszystkimi powiązanymi z nim zasobami. Otwórz selektor projektów
- Sprawdź, czy w projekcie Cloud włączone są płatności. Dowiedz się, jak sprawdzić, czy w projekcie są włączone płatności.
- włączyć Compute Engine, Vertex API, Notebook API i Vision AI API; Włączanie interfejsów API
Utwórz konto usługi:
- W konsoli Google Cloud otwórz stronę Utwórz konto usługi. Otwórz stronę tworzenia konta usługi.
- Wybierz projekt.
- W polu Nazwa konta usługi wpisz nazwę. Konsola Google Cloud wypełnia pole Identyfikator konta usługi na podstawie tej nazwy. W polu Opis konta usługi wpisz opis. Przykład: Konto usługi do krótkiego wprowadzenia.
- Kliknij Utwórz i kontynuuj.
- Aby zapewnić dostęp do projektu, przypisz do konta usługi te role:
- Vision AI > Edytujący Vision AI
- Compute Engine > Administrator instancji Compute (beta)
- BigQuery > Administrator BigQuery .
Na liście Wybierz rolę wybierz rolę. Aby dodać kolejne role, kliknij Dodaj kolejną rolę i dodaj każdą z nich.
- Kliknij Dalej.
- Aby zakończyć tworzenie konta usługi, kliknij Gotowe. Nie zamykaj okna przeglądarki. Użyjesz go w następnym kroku.
3. Konfigurowanie notatnika Jupyter
Zanim utworzysz aplikację w ramach analizy zajętości, musisz zarejestrować strumień, którego aplikacja może później używać.
W tym samouczku utworzysz instancję notatnika Jupyter, która hostuje film, i wyślesz te strumieniowane dane wideo z notatnika. Korzystamy z notatnika Jupyter, ponieważ daje on nam elastyczność wykonywania poleceń powłoki oraz uruchamiania niestandardowego kodu przed i po przetworzeniu w jednym miejscu, co pozwala na szybkie eksperymentowanie. Będziemy używać tego notatnika do:
- Uruchom serwer rtsp jako proces w tle
- Uruchom polecenie vaictl jako proces w tle
- Uruchamiaj zapytania i kod przetwarzania, aby analizować dane wyjściowe analizy zajętości
Tworzenie notatnika Jupyter
Pierwszym krokiem podczas wysyłania filmu z instancji notatnika Jupyter jest utworzenie notatnika z naszym kontem usługi utworzonym w poprzednim kroku.
- W konsoli otwórz stronę Vertex AI. Otwórz Vertex AI Workbench
- Kliknij Notatniki zarządzane przez użytkownika.
- Kliknij Nowy notatnik > Tensorflow Enterprise 2.6 (z LTS) > Bez GPU
- Wpisz nazwę notatnika Jupyter. Więcej informacji znajdziesz w artykule Konwencja nazewnictwa zasobów.
- Kliknij OPCJE ZAAWANSOWANE.
- Przewiń w dół do sekcji Sekcje uprawnień.
- Odznacz opcję Użyj domyślnego konta usługi Compute Engine.
- Dodaj adres e-mail konta usługi utworzony w poprzednim kroku. Następnie kliknij Utwórz.
- Po utworzeniu instancji kliknij OTWÓRZ JUPYTERLAB.
4. Konfigurowanie notatnika do strumieniowego przesyłania wideo
Zanim utworzysz aplikację w ramach analizy zajętości, musisz zarejestrować strumień, którego aplikacja może później używać.
W tym samouczku użyjemy naszej instancji notatnika Jupyter do hostowania filmu, a Ty wyślesz strumieniowane dane wideo z terminala notatnika.
Pobieranie narzędzia wiersza poleceń vaictl
- W otwartej instancji JupyterLab otwórz notatnik z programu uruchamiającego.
- Pobierz narzędzie wiersza poleceń Vertex AI Vision (vaictl), narzędzie wiersza poleceń serwera rtsp i narzędzie open-cv, korzystając z następującego polecenia w komórce notatnika:
!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. Pozyskaj plik wideo na potrzeby przesyłania strumieniowego
Po skonfigurowaniu wymaganych narzędzi wiersza poleceń w środowisku notatnika możesz skopiować przykładowy plik wideo, a następnie za pomocą vaictl przesłać dane wideo do aplikacji do analizy zajętości.
Rejestrowanie nowej transmisji
- Kliknij kartę strumienie w lewym panelu Vertex AI Vision.
- Kliknij przycisk Zarejestruj u góry strony
- W nazwie strumienia wpisz „queue-stream”.
- W regionie wybierz ten sam region, który został wybrany podczas tworzenia notatnika w poprzednim kroku.
- Kliknij Zarejestruj.
Kopiowanie przykładowego filmu do maszyny wirtualnej
- Skopiuj w notatniku przykładowy film, korzystając z podanego niżej polecenia wget.
!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4
Przesyłanie strumieniowe filmów z maszyny wirtualnej i pozyskiwanie danych do strumienia
- Aby wysłać ten lokalny plik wideo do strumienia danych wejściowych aplikacji, użyj tego polecenia w komórce notatnika. Musisz zastąpić te zmienne:
- PROJECT_ID: identyfikator Twojego projektu Google Cloud.
- LOCATION: identyfikator lokalizacji. Na przykład us-central1. Więcej informacji znajdziesz w artykule Lokalizacje Cloud.
- LOCAL_FILE: nazwa lokalnego pliku wideo. np.
seq25_h264.mp4.
PROJECT_ID='<Your Google Cloud project ID>'
LOCATION='<Your stream location>'
LOCAL_FILE='seq25_h264.mp4'
STREAM_NAME='queue-stream'
- Uruchomienie serwera rtsp-simple-server, na którym będziemy przesyłać strumieniowo plik wideo z użyciem protokołu 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)
- Użyj narzędzia wiersza poleceń ffmpeg, aby zapętlić film w strumieniu 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)
- Użyj narzędzia wiersza poleceń vaictl, aby przesłać strumieniowo film z identyfikatora URI serwera rtsp do strumienia „queue-stream” w Vertex AI Vision utworzony w poprzednim kroku.
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)
Od rozpoczęcia procesu przetwarzania vaictl do wyświetlenia filmu w panelu może upłynąć około 100 sekund.
Gdy przetwarzanie strumienia będzie dostępne, zobaczysz kanał wideo na karcie Strumienie w panelu Vertex AI Vision, wybierając strumień z kolejki.
6. Tworzenie aplikacji
Pierwszym krokiem jest utworzenie aplikacji, która będzie przetwarzać Twoje dane. Aplikację można traktować jako automatyczny potok łączący:
- Przetwarzanie danych: kanał wideo jest przetwarzany w strumieniu.
- Analiza danych: model AI(Computer Vision) można dodać po przetworzeniu danych.
- Przechowywanie danych: 2 wersje kanału wideo (pierwotny i strumień przetwarzany przez model AI) mogą być przechowywane w hurtowni multimediów.
W konsoli Google Cloud aplikacja ma postać wykresu.
Tworzenie pustej aplikacji
Zanim wypełnisz wykres aplikacji, musisz utworzyć pustą aplikację.
utworzyć aplikację w konsoli Google Cloud,
- Otwórz konsolę Google Cloud.
- Otwórz kartę Applications (Aplikacje) w panelu Vertex AI Vision. Otwórz kartę Aplikacje.
- Kliknij przycisk Utwórz.
- Wpisz „queue-app” jako nazwę aplikacji i wybierz region.
- Kliknij Utwórz.
Dodawanie węzłów komponentów aplikacji
Po utworzeniu pustej aplikacji możesz dodać te 3 węzły do wykresu aplikacji:
- Węzeł przetwarzania: zasób strumienia, który pozyskuje dane wysyłane z serwera wideo rtsp utworzonego w notatniku.
- Węzeł przetwarzania: model analizy zajętości, który działa na pozyskanych danych.
- Węzeł pamięci masowej: hurtownia multimediów, w której przechowywane są przetworzone filmy, która służy jako magazyn metadanych. Magazyny metadanych obejmują informacje analityczne o przetworzonych danych wideo oraz informacje wywnioskowane przez modele AI.
Dodaj węzły komponentów do swojej aplikacji w konsoli.
- Otwórz kartę Applications (Aplikacje) w panelu Vertex AI Vision. Otwórz kartę Aplikacje.
Wyświetli się wizualizacja wykresu potoku przetwarzania.
Dodawanie węzła pozyskiwania danych
- Aby dodać węzeł strumienia wejściowego, wybierz opcję Strumienie w sekcji Oprogramowanie sprzęgające menu bocznego.
- W sekcji Źródło menu Strumień, które się otworzy, wybierz Dodaj strumienie.
- W menu Dodaj strumienie wybierz kolejka.
- Aby dodać strumień do wykresu aplikacji, kliknij Dodaj strumienie.
Dodawanie węzła przetwarzania danych
- Aby dodać węzeł modelu liczby obłożenia, wybierz opcję Analiza zajętości w sekcji Modele specjalistyczne w menu bocznym.
- Pozostaw domyślne ustawienie Osoby. Odznacz opcję Pojazdy, jeśli jest już wybrana.
- W sekcji Advanced Options (Opcje zaawansowane) kliknij Create Active Zones/Lines (Utwórz aktywne strefy/linie)
. - Narysuj aktywne strefy za pomocą narzędzia Wielokąt, aby policzyć osoby znajdujące się w danej strefie. Odpowiednio oznacz strefę
- Kliknij u góry strzałkę Wstecz.
- Kliknij pole wyboru, aby dodać ustawienia czasu kontaktu, aby wykrywać zator.
Dodawanie węzła przechowywania danych
- Aby dodać węzeł miejsca docelowego danych wyjściowych (pamięć masowa), wybierz opcję VIsion AI Warehouse w sekcji Connectors (Łączniki) w bocznym menu.
- Kliknij oprogramowanie sprzęgające Vertex AI Warehouse, aby otworzyć jego menu, a następnie kliknij Połącz hurtownię.
- W menu Połącz hurtownię wybierz Utwórz nową hurtownię. Nazwij magazyn queue-warehouse i pozostaw wartość TTL po 14 dniach.
- Kliknij przycisk Utwórz, aby dodać hurtownię.
7. Połącz dane wyjściowe z tabelą BigQuery
Gdy dodasz oprogramowanie sprzęgające BigQuery do aplikacji Vertex AI Vision, wszystkie dane wyjściowe połączonego modelu aplikacji będą przetwarzane do tabeli docelowej.
Możesz utworzyć własną tabelę BigQuery i określić ją podczas dodawania oprogramowania sprzęgającego BigQuery do aplikacji lub pozwolić, aby platforma aplikacji Vertex AI Vision została utworzona automatycznie.
Automatyczne tworzenie tabeli
Jeśli zezwalasz platformie aplikacji Vertex AI Vision na automatyczne utworzenie tabeli, możesz określić tę opcję podczas dodawania węzła oprogramowania sprzęgającego BigQuery.
Jeśli chcesz korzystać z automatycznego tworzenia tabel, obowiązują te warunki zbioru danych i tabeli:
- Zbiór danych: nazwa zbioru danych tworzonego automatycznie to visionai_dataset.
- Tabela: nazwa automatycznie tworzonej tabeli to visionai_dataset.APPLICATION_ID.
- Obsługa błędów:
- Jeśli w ramach tego samego zbioru danych istnieje tabela o tej samej nazwie, nie następuje automatyczne tworzenie.
- Otwórz kartę Applications (Aplikacje) w panelu Vertex AI Vision. Otwórz kartę Aplikacje.
- Wybierz Wyświetl aplikację obok nazwy aplikacji na liście.
- Na stronie kreatora aplikacji w sekcji Oprogramowanie sprzęgające wybierz BigQuery.
- Pole Ścieżka BigQuery pozostaw puste.
- W metadanych sklepu z wybierz tylko „Analiza zajętości” i odznacz strumienie.
Końcowy wykres aplikacji powinien wyglądać tak:
8. Wdrażanie aplikacji do użytkowania
Gdy stworzysz kompleksową aplikację ze wszystkimi niezbędnymi komponentami, ostatnim krokiem będzie jej wdrożenie.
- Otwórz kartę Applications (Aplikacje) w panelu Vertex AI Vision. Otwórz kartę Aplikacje.
- Wybierz Wyświetl aplikację obok pozycji aplikacji do kolejki na liście.
- Na stronie Studio kliknij przycisk Wdróż.
- W oknie potwierdzenia kliknij Wdróż. Wdrażanie może potrwać kilka minut. Po zakończeniu wdrażania obok węzłów pojawią się zielone znaczniki wyboru.
9. Wyszukaj treści wideo w hurtowni miejsca na dane
Po pozyskaniu danych wideo do aplikacji do przetwarzania danych możesz wyświetlać przeanalizowane dane wideo i przeszukiwać je na podstawie informacji o obłożeniu.
- Otwórz kartę Magazyny w panelu Vertex AI Vision. Otwórz kartę Hurtownie
- Znajdź na liście magazyn magazynu kolejek i kliknij Wyświetl zasoby.
- W sekcji Liczba osób ustaw wartość Min na 1, a wartość Maks. na 5.
- Aby filtrować przetworzone dane wideo przechowywane w hurtowni multimediów Vertex AI Vision, kliknij Szukaj.
Widok przechowywanych danych wideo zgodnych z kryteriami wyszukiwania w konsoli Google Cloud.
10. Dodawanie adnotacji do danych wyjściowych i analizowanie ich za pomocą tabeli BigQuery
- W notatniku zainicjuj w komórce poniższe zmienne.
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'
- Teraz przechwytujemy klatki ze strumienia rtsp za pomocą następującego kodu:
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')
- Pobierz sygnaturę czasową danych i informacje o adnotacji z tabeli BigQuery i utwórz katalog do przechowywania przechwyconych obrazów ramek:
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)
- Dodaj adnotacje do ramek przy użyciu tego kodu:
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')
- Zatrzymaj zadanie adnotacji, klikając przycisk Zatrzymaj na pasku menu notatnika.
- Możesz sprawdzać poszczególne klatki, korzystając z tego kodu:
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. Gratulacje
Gratulacje! Moduł ukończony.
Czyszczenie danych
Aby uniknąć obciążenia konta Google Cloud opłatami za zasoby zużyte w tym samouczku, możesz usunąć projekt zawierający te zasoby lub zachować projekt i usunąć poszczególne zasoby.
Usuwanie projektu
Usuwanie poszczególnych zasobów
Materiały
https://cloud.google.com/vision-ai/docs/overview
https://cloud.google.com/vision-ai/docs/occupancy-count-tutorial