1. 목표
개요
이 Codelab에서는 소매업 동영상 영상을 사용하여 대기열 크기를 모니터링하는 Vertex AI Vision 애플리케이션을 엔드 투 엔드로 만드는 데 중점을 둡니다. 선행 학습된 특수 모델인 숙박 인원 분석에 내장된 기능을 사용하여 다음 정보를 파악합니다.
- 대기열에 서 있는 사람의 수를 셉니다.
- 카운터에서 서빙되는 사람의 수를 계산합니다.
학습할 내용
- Vertex AI Vision에서 애플리케이션을 만들고 배포하는 방법
- 동영상 파일을 사용하여 RTSP 스트림을 설정하고 Jupyter 노트북의 vaictl을 사용하여 스트림을 Vertex AI Vision으로 수집하는 방법
- 점유율 분석 모델 및 다양한 기능을 사용하는 방법
- 스토리지 Vertex AI Vision의 미디어 웨어하우스에서 동영상을 검색하는 방법입니다.
- 출력을 BigQuery에 연결하고, SQL 쿼리를 작성하여 모델의 json 출력에서 유용한 정보를 추출하고, 출력을 사용해 원본 동영상에 라벨을 지정하고 주석을 추가하는 방법
비용:
Google Cloud에서 이 실습을 진행하는 데 드는 총 비용은 약 $2입니다.
2. 시작하기 전에
프로젝트를 만들고 API를 사용 설정합니다.
- Google Cloud 콘솔의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다. 참고: 이 절차에서 생성한 리소스를 유지하지 않으려면 기존 프로젝트를 선택하지 말고 프로젝트를 새로 만드세요. 이러한 단계가 완료되면 프로젝트를 삭제하여 프로젝트와 연결된 모든 리소스를 삭제할 수 있습니다. 프로젝트 선택기로 이동
- Cloud 프로젝트에 결제가 사용 설정되어 있어야 하므로 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.
- Compute Engine, Vertex API, Notebook API, Vision AI API를 사용 설정합니다. API 사용 설정
서비스 계정 만들기:
- Google Cloud Console에서 서비스 계정 만들기 페이지로 이동합니다. 서비스 계정 만들기로 이동
- 프로젝트를 선택합니다.
- 서비스 계정 이름 필드에 이름을 입력합니다. Google Cloud 콘솔에서 이 이름을 기준으로 서비스 계정 ID 필드를 채웁니다. 서비스 계정 설명 필드에 설명을 입력합니다. 예를 들어 빠른 시작의 서비스 계정입니다.
- 만들고 계속하기를 클릭합니다.
- 프로젝트에 대한 액세스 권한을 제공하려면 서비스 계정에 다음 역할을 부여합니다.
- Vision AI > Vision AI 편집자
- Compute Engine> Compute 인스턴스 관리자 (베타)
- BigQuery> BigQuery 관리자
역할 선택 목록에서 역할을 선택합니다. 역할을 추가하려면 다른 역할 추가를 클릭하고 역할을 각각 추가합니다.
- 계속을 클릭합니다.
- 완료를 클릭하여 서비스 계정 만들기를 마칩니다. 브라우저 창을 닫지 마세요. 다음 단계에서 사용합니다.
3. Jupyter 노트북 설정
점유율 분석에서 앱을 만들기 전에 나중에 앱에서 사용할 수 있는 스트림을 등록해야 합니다.
이 튜토리얼에서는 동영상을 호스팅하는 Jupyter 노트북 인스턴스를 만들고 노트북에서 이 스트리밍 동영상 데이터를 전송합니다. jupyter 노트북을 사용하는 이유는 셸 명령어를 실행할 뿐만 아니라 커스텀 전/후 처리 코드를 한곳에서 실행할 수 있는 유연성을 제공하므로 신속한 실험에 매우 적합하기 때문입니다. 이 노트북은 다음 용도로 사용됩니다.
- rtsp 서버를 백그라운드 프로세스로 실행
- vaictl 명령어를 백그라운드 프로세스로 실행
- 쿼리를 실행하고 코드를 처리하여 점유율 분석 결과를 분석합니다.
Jupyter 노트북 만들기
Jupyter 노트북 인스턴스에서 동영상을 전송하는 첫 번째 단계는 이전 단계에서 만든 서비스 계정으로 노트북을 만드는 것입니다.
- 콘솔에서 Vertex AI 페이지로 이동합니다. Vertex AI Workbench로 이동
- 사용자 관리 노트북을 클릭합니다.
- 새 노트북 > Tensorflow Enterprise 2.6 (LTS 포함) > GPU 사용 안 함
- jupyter 노트북의 이름을 입력합니다. 자세한 내용은 리소스 이름 지정 규칙을 참고하세요.
- 고급 옵션을 클릭합니다.
- 권한 섹션까지 아래로 스크롤합니다.
- Compute Engine 기본 서비스 계정 사용 옵션을 선택 해제합니다.
- 이전 단계에서 만든 서비스 계정 이메일을 추가합니다. 만들기를 클릭합니다.
- 인스턴스가 생성되면 JupyterLab 열기를 클릭합니다.
4. 동영상을 스트리밍하기 위한 노트북 설정
점유율 분석에서 앱을 만들기 전에 나중에 앱에서 사용할 수 있는 스트림을 등록해야 합니다.
이 튜토리얼에서는 Jupyter 노트북 인스턴스를 사용하여 동영상을 호스팅하고 해당 스트리밍 동영상 데이터를 노트북 터미널에서 전송합니다.
vaictl 명령줄 도구 다운로드
- 열려 있는 Jupyterlab 인스턴스의 런처에서 노트북을 엽니다.
- 노트북 셀에서 다음 명령어를 사용하여 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을 사용하여 동영상 데이터를 점유율 분석 앱으로 스트리밍할 수 있습니다.
새 스트림 등록
- Vertex AI Vision의 왼쪽 패널에서 스트림 탭을 클릭합니다.
- 상단의
버튼을 클릭합니다. - 스트림 이름에 'queue-stream'을 입력합니다.
- 리전 내에서 이전 단계의 노트북 생성 중에 선택한 것과 동일한 리전을 선택합니다.
- 등록을 클릭합니다.
VM에 샘플 동영상 복사하기
- 노트북에서 다음 wget 명령어를 사용하여 샘플 동영상을 복사합니다.
!wget -q https://github.com/vagrantism/interesting-datasets/raw/main/video/collective_activity/seq25_h264.mp4
VM에서 동영상을 스트리밍하고 스트림으로 데이터를 수집합니다.
- 이 로컬 동영상 파일을 앱 입력 스트림으로 전송하려면 노트북 셀에서 다음 명령어를 사용합니다. 다음과 같이 대체해야 합니다.
- 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'
- rtsp 프로토콜로 동영상 파일을 스트리밍하는 rtsp-simple-server 시작
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)
- 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)
- 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 대시보드의 스트림 탭에서 동영상 피드를 볼 수 있습니다.
6. 애플리케이션 만들기
첫 번째 단계는 데이터를 처리하는 앱을 만드는 것입니다. 앱은 다음을 연결하는 자동화된 파이프라인으로 생각할 수 있습니다.
- 데이터 수집: 동영상 피드가 스트림으로 처리됩니다.
- 데이터 분석: 수집 후 AI(컴퓨터 비전) 모델을 추가할 수 있습니다.
- 데이터 저장: 두 가지 버전의 동영상 피드 (원본 스트림과 AI 모델에서 처리한 스트림)를 미디어 웨어하우스에 저장할 수 있습니다.
Google Cloud 콘솔에서 앱은 그래프로 표시됩니다.
빈 앱 만들기
앱 그래프를 채우려면 먼저 빈 앱을 만들어야 합니다.
Google Cloud 콘솔에서 앱을 만듭니다.
- Google Cloud 콘솔로 이동합니다.
- Vertex AI Vision 대시보드의 애플리케이션 탭을 엽니다. 애플리케이션 탭으로 이동
- 만들기 버튼을 클릭합니다.
- 앱 이름으로 ‘queue-app'을 입력하고 리전을 선택합니다.
- 만들기를 클릭합니다.
앱 구성요소 노드 추가
빈 애플리케이션을 만든 후에는 앱 그래프에 3개의 노드를 추가할 수 있습니다.
- 수집 노드: 노트북에서 만든 RTSP 동영상 서버에서 전송된 데이터를 수집하는 스트림 리소스입니다.
- 처리 노드: 수집된 데이터에 대해 작동하는 점유율 분석 모델입니다.
- 저장소 노드: 처리된 동영상을 저장하고 메타데이터 저장소 역할을 하는 미디어 웨어하우스. 메타데이터 저장소에는 수집된 동영상 데이터에 대한 분석 정보와 AI 모델에서 추론한 정보가 포함됩니다.
콘솔에서 앱에 구성요소 노드를 추가합니다.
- Vertex AI Vision 대시보드의 애플리케이션 탭을 엽니다. 애플리케이션 탭으로 이동
그러면 처리 파이프라인의 그래프 시각화로 이동합니다.
데이터 수집 노드 추가
- 입력 스트림 노드를 추가하려면 사이드 메뉴의 커넥터 섹션에서 스트림 옵션을 선택합니다.
- 열리는 스트림 메뉴의 소스 섹션에서 스트림 추가를 선택합니다.
- 스트림 추가 메뉴에서 queue-stream을 선택합니다.
- 앱 그래프에 스트림을 추가하려면 스트림 추가를 클릭합니다.
데이터 처리 노드 추가
- 점유율 모델 노드를 추가하려면 사이드 메뉴의 특수 모델 섹션에서 숙박 인원 분석 옵션을 선택합니다.
- 기본 선택 항목인 사용자는 그대로 둡니다. 차량이 이미 선택되어 있는 경우 선택 해제합니다.
- 고급 옵션 섹션에서 활성 영역/선 만들기 를 클릭합니다.
- 다각형 도구를 사용해 활성 영역을 그려 해당 구역의 인원수를 집계합니다. 영역에 적절하게 라벨 지정
- 상단의 뒤로 화살표를 클릭합니다.
- 체크박스를 클릭하여 체류 시간 설정을 추가하여 정체를 감지합니다.
데이터 스토리지 노드 추가
- 출력 대상 (스토리지) 노드를 추가하려면 사이드 메뉴의 커넥터 섹션에서 VIsion AI Warehouse 옵션을 선택합니다.
- Vertex AI Warehouse 커넥터를 클릭하여 메뉴를 열고 웨어하우스 연결을 클릭합니다.
- 웨어하우스 연결 메뉴에서 새 웨어하우스 만들기를 선택합니다. 웨어하우스의 이름을 queue-warehouse로 지정하고 TTL 기간은 14일로 둡니다.
- 만들기 버튼을 클릭하여 웨어하우스를 추가합니다.
7. BigQuery 테이블에 출력 연결
BigQuery 커넥터를 Vertex AI Vision 앱에 추가하면 연결된 모든 앱 모델 출력이 대상 테이블로 수집됩니다.
자체 BigQuery 테이블을 만들고 앱에 BigQuery 커넥터를 추가할 때 이 테이블을 지정하거나 Vertex AI Vision 앱 플랫폼에서 자동으로 테이블을 만들도록 할 수 있습니다.
자동 테이블 생성
Vertex AI Vision 앱 플랫폼에서 테이블을 자동으로 만들도록 하면 BigQuery 커넥터 노드를 추가할 때 이 옵션을 지정할 수 있습니다.
자동 테이블 생성을 사용하려면 다음 데이터 세트 및 테이블 조건이 적용됩니다.
- 데이터 세트: 자동으로 생성되는 데이터 세트 이름은 visionai_dataset입니다.
- 테이블: 자동으로 생성되는 테이블 이름은 visionai_dataset.APPLICATION_ID입니다.
- 오류 처리:
- 동일한 데이터 세트에 동일한 이름의 테이블이 있는 경우 자동으로 생성되지 않습니다.
- Vertex AI Vision 대시보드의 애플리케이션 탭을 엽니다. 애플리케이션 탭으로 이동
- 목록에서 애플리케이션 이름 옆에 있는 앱 보기를 선택합니다.
- 애플리케이션 빌더 페이지의 커넥터 섹션에서 BigQuery를 선택합니다.
- BigQuery path 필드는 비워 둡니다.
- 스토어 메타데이터 출처: ‘숙박 인원 분석’만 선택하고 스트림을 선택 해제합니다.
최종 앱 그래프는 다음과 같이 표시됩니다.
8. 사용을 위해 앱 배포하기
필요한 모든 구성요소를 갖춘 엔드 투 엔드 앱을 빌드한 후 앱을 사용하기 위한 마지막 단계는 앱을 배포하는 것입니다.
- Vertex AI Vision 대시보드의 애플리케이션 탭을 엽니다. 애플리케이션 탭으로 이동
- 목록에서 queue-app 앱 옆에 있는 앱 보기를 선택합니다.
- 스튜디오 페이지에서 배포 버튼을 클릭합니다.
- 다음 확인 대화상자에서 배포를 클릭합니다. 배포 작업을 완료하는 데 몇 분 정도 걸릴 수 있습니다. 배포가 완료되면 노드 옆에 녹색 체크표시가 나타납니다.
9. 스토리지 웨어하우스에서 동영상 콘텐츠 검색
동영상 데이터를 처리 앱으로 수집한 후에는 분석된 동영상 데이터를 확인하고 점유율 분석 정보를 기반으로 데이터를 검색할 수 있습니다.
- Vertex AI Vision 대시보드의 웨어하우스 탭을 엽니다. 웨어하우스 탭으로 이동
- 목록에서 대기열 창고 웨어하우스를 찾고 애셋 보기를 클릭합니다.
- 인물 수 섹션에서 최소 값을 1로, 최대 값을 5로 설정합니다.
- Vertex AI Vision의 미디어 웨어하우스에 저장된 처리된 동영상 데이터를 필터링하려면 검색을 클릭합니다.
Google Cloud 콘솔의 검색 기준과 일치하는 저장된 동영상 데이터의 뷰
10. BigQuery 테이블을 사용하여 출력 주석 달기 및 분석하기
- 노트북에서 셀의 다음 변수를 초기화합니다.
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'
- 이제 다음 코드를 사용하여 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')
- 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)
- 다음 코드를 사용하여 프레임에 주석을 추가합니다.
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')
- 노트북 메뉴 바의 중지 버튼을 사용하여 주석 작업을 중지합니다.
- 다음 코드를 사용하여 개별 프레임을 다시 방문할 수 있습니다.
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. 축하합니다
축하합니다. 실습을 완료하셨습니다.
정리
이 튜토리얼에서 사용된 리소스 비용이 Google Cloud 계정에 청구되지 않도록 하려면 리소스가 포함된 프로젝트를 삭제하거나 프로젝트를 유지하고 개별 리소스를 삭제하세요.
프로젝트 삭제
개별 리소스 삭제
리소스
https://cloud.google.com/vision-ai/docs/overview
https://cloud.google.com/vision-ai/docs/occupancy-count-tutorial