이 페이지는 Cloud Translation API를 통해 번역되었습니다.

Vertex AI Vision 큐 감지 앱

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 콘솔에서 서비스 계정 만들기 페이지로 이동합니다. 서비스 계정 만들기로 이동
프로젝트를 선택합니다.
서비스 계정 이름 필드에 이름을 입력합니다. Google Cloud 콘솔은 이 이름을 기반으로 서비스 계정 ID 필드를 채웁니다. 서비스 계정 설명 입력란에 설명을 입력합니다. 예를 들어 빠른 시작의 서비스 계정입니다.
만들고 계속하기를 클릭합니다.
프로젝트에 대한 액세스 권한을 제공하려면 서비스 계정에 다음 역할을 부여합니다.

Vision AI > Vision AI 편집기
Compute Engine > Compute 인스턴스 관리자 (베타)
BigQuery > BigQuery 관리자

역할 선택 목록에서 역할을 선택합니다. 역할을 추가하려면 다른 역할 추가를 클릭하고 각 역할을 추가합니다.

계속을 클릭합니다.
완료를 클릭하여 서비스 계정 만들기를 마칩니다. 브라우저 창을 닫지 마세요. 다음 단계에서 사용합니다.

3. Jupyter 노트북 설정

점유율 분석에서 앱을 만들기 전에 나중에 앱에서 사용할 수 있는 스트림을 등록해야 합니다.

이 튜토리얼에서는 동영상을 호스팅하는 Jupyter 노트북 인스턴스를 만들고 노트북에서 스트리밍 동영상 데이터를 전송합니다. Jupyter 노트북은 셸 명령어를 실행하고 맞춤 전처리/후처리 코드를 한곳에서 실행할 수 있는 유연성을 제공하므로 빠른 실험에 매우 적합합니다. 이 노트북에서는 다음 작업을 수행합니다.

rtsp 서버를 백그라운드 프로세스로 실행
vaictl 명령어를 백그라운드 프로세스로 실행
쿼리 실행 및 코드 처리로 숙박 인원 분석 출력 분석

Jupyter 노트북 만들기

Jupyter Notebook 인스턴스에서 동영상을 전송하는 첫 번째 단계는 이전 단계에서 만든 서비스 계정으로 노트북을 만드는 것입니다.

콘솔에서 Vertex AI 페이지로 이동합니다. Vertex AI Workbench로 이동
'사용자 관리 노트북'을 클릭합니다.

새 노트북 > Tensorflow Enterprise 2.6 (LTS 사용) > GPU 사용 안 함을 클릭합니다.

Jupyter 노트북의 이름을 입력합니다. 자세한 내용은 리소스 이름 지정 규칙을 참고하세요.

고급 옵션을 클릭합니다.
권한 섹션으로 아래로 스크롤합니다.
Compute Engine 기본 서비스 계정 사용 옵션을 선택 해제합니다.
이전 단계에서 만든 서비스 계정 이메일을 추가합니다. 만들기를 클릭합니다.

인스턴스가 생성되면 JupyterLab 열기를 클릭합니다.

4. 동영상을 스트리밍하도록 노트북 설정하기

점유 분석에서 앱을 만들기 전에 나중에 앱에서 사용할 수 있는 스트림을 등록해야 합니다.

이 튜토리얼에서는 Jupyter 노트북 인스턴스를 사용하여 동영상을 호스팅하고 노트북 터미널에서 스트리밍 동영상 데이터를 전송합니다.

vaictl 명령줄 도구 다운로드

열린 Jupyterlab 인스턴스에서 런처에서 Notebook을 엽니다.

노트북 셀에서 다음 명령어를 사용하여 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 대시보드의 스트림 탭에서 queue-stream 스트림을 선택하여 동영상 피드를 볼 수 있습니다.

스트림 탭으로 이동

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 테이블에 출력 연결

Vertex AI Vision 앱에 BigQuery 커넥터를 추가하면 연결된 모든 앱 모델 출력이 대상 테이블에 처리됩니다.

자체 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 대시보드의 창고 탭을 엽니다. 창고 탭으로 이동
목록에서 queue-warehouse 창고를 찾아 확장 소재 보기를 클릭합니다.
인원 수 섹션에서 최소 값을 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>&nbsp;</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

라이선스

설문조사

이 튜토리얼을 어떻게 사용하셨나요?

읽기만 할 계획입니다

읽은 다음 연습 활동을 완료했습니다

이 Codelab이 얼마나 유용했나요?

매우 유용함

다소 유용함

유용하지 않음

이 Codelab을 따라 진행하기는 얼마나 쉬웠나요?

쉬움

보통

어려움