📹

Vision Studio で簡単リアルタイム映像解析アプリ

Yuichiro Danno

2023/12/16に公開

こんにちは、メディア・エンターテイメント業界担当の Customer Engineer の Dan です。

この記事は Google Cloud Japan Advent Calendar 2023 (通常版) の 16 日目の記事です。

本記事では、全然知られていない（ググっても日本語の記事がまったくない） Vertex AI Vision Studio というサービスについてご紹介したいと思います。紛らわしいですが、画像認識サービス Cloud Vision API や生成 AI モデル・アプリ開発ツールGenerative AI Studio とも違います。

リアルタイム映像解析の世界

世は生成 AI/LLM 全盛の時代ですが、まだまだ従来型の事前学習済みモデルの AI ソリューションの出番は多いです。特にリアルタイムに映像解析を行い、オブジェクト検知してアプリケーションへ通知するようなリアルタイム映像解析の世界（*）では、求められるレイテンシの低さや認識・識別の精度の高さから事前学習済みモデルが採用されています。
*例）店舗や駐車場の混雑度、レジやゲートでの待機列、商品の棚在庫管理や棚ごとの立ち寄り数の計測、入出店者数のカウントなど

とはいえ、リアルタイム映像解析アプリケーションの構築は、モデルの構築、映像処理、アプリケーションへの連携など、専門人材でないと中々難しい場面が多いです。
事前学習済みモデルを使って、イベント発火条件とか定義できて、簡単にアプリ連携できると最高ですよね。しかもコード書かずにできるソリューションなんてあった日にはもう使うしかないですね。

そんな便利なソリューションが Google Cloud にはあるのです。
その名も Vertex AI Vision Studio（以下、Vision Studio）。Google Cloud の機械学習プラットフォーム Vertex AI 上のマネージドサービスです。

今回はサンプルアプリの構築を通じて、Vision Studio の魅力に触れていきます。

Vertex AI Vision Studio とは

Vertex AI Vision Studio は、ストリーミング・動画ファイル等の映像ファイルを取り込み分析し処理を行う End-to-End の画像認識アプリケーションを構築可能な、フルマネージドのブラウザ開発環境です。
なお、アプリの宛先としてVision Warehouse を利用することで、映像データの保存および格納後の高度なメディア検索（Embeddings を活用したマルチモーダル検索）も行うことが可能です。出版社・新聞社・メディア等の社内アセット検索に有用な、単体でも利用可能なソリューションですので、別の機会に改めてご紹介したいと思います。

主要な機能や利点

ブラウザ開発環境

Vision Studio は、開発環境としてはブラウザ上でコンポーネントを選択し設定するだけのシンプルなインターフェースを提供しており、これによりユーザーはアプリを迅速に構築してデプロイすることが可能です。

利用可能な事前学習済みモデル

現時点でサポートされている事前学習済みのモデルは下記です。

一般モデル
- Person/vehicle detector（人・車の検出）
- Person blur（人に対するモザイク処理）
- Object detector（オブジェクト検出）
- PPE detector（Personal Protect Equipment[個人用保護具] : ヘルメット、フェイスマスクなどの検出）
特定業界向けモデル
- Occupancy analytics
- Tag recognizer ※独自学習済みモデル（AutoMLモデル）が必要
- Product recognizer ※独自学習済みモデル（AutoMLモデル）が必要

これら以外にも AutoML で学習済みのカスタムモデルの利用も可能です。

ユースケース

Vision Studio および事前学習済みモデルを活用することで、下記のようなユースケースに対応したアプリを簡単に構築可能です。

工場での製品認識、店舗での棚札認識
店舗や駐車場、イベントでの混雑度計測
レジやゲートでの待機列、入出店者数のカウント

高度な機能

でもちょっと待って下さい。
事前学習済みモデルを使えるのは分かったけど、それだけでは映像解析アプリにはなり得ないのでは？と思いませんか。例えば入出店者数のカウントをしたい場合、映像の中のどのラインを超えた場合にカウントするとか条件が必要で、しかも解析したいシーン・映像によって条件が異なるから、結構難しいのではと思いますよね。
そういう条件を指定できる機能として「Active zone and line editor」があります。

例えば、Active Line 機能を使うと、映像内でこの線を横切った対象（例：人・車）をカウントする、といった境界ラインと条件を指定することができます。

また、Active Zone 機能を使うと、映像内でこの領域（ゾーン）内に入っている対象（例：人・車）をカウントする、といった境界ゾーンと条件を指定することができます。

このような条件指定がブラウザで指定できるって、めちゃすごくないですか？
でも本当に想定通り動作するのか、本当に簡単に作れるのか、試してみたいと思います。

リアルタイム映像解析アプリを作ってみる

今回はちょうど店舗内レジの待機列のサンプル映像があったので、レジの待機列検出をしてみたいと思います。
例えばコンビニやカフェ等の小規模店舗の店員さんは、品出しからレジまで一人で行う業務の幅が広いため、他の作業もしつつ混雑してきたらレジをオープンして対応しないといけないと思います。そういうときに、レジの待機列が X 人を超えたら通知、みたいなアプリがあると便利ですよね。

イメージとしては、下図のような感じです。実際の実装だと、通知部分は Firebase でスマホアプリにプッシュ通知とかが考えられますが、今回は Cloud Functions で受け取るところまで確認します。
また、あとで混雑状況の分析のために、BigQuery にもデータを貯めようと思います。

事前準備

映像インジェストサーバ環境構築（今回は Cloud Shell）

こちらのステップは、映像ファイルを GCS にアップロードして映像解析したい場合は不要です。
この作業は、今回の目的のようにリアルタイム映像解析をしたい場合、カメラ映像を Vision Studio に送るためのサーバー側で Google Cloud および Vision Studio 用の CLI ツール (vaictl) の設定が必要であるため、行っております。前述の画像内だと、映像ストリーム送信サーバ（Compute Engine）での作業です。

事前準備は、Vertex AI Vision Studio - Set up a project and a development environment に従って行います。当記事の本筋ではないので、説明は割愛します。
一点注意点としまして、vaictl の利用は OS: Debian GNU/Linux, CPU architecture: x86_64 の制限がある旨、ご注意下さい。今回は手早くテストするため、Cloud Shell 上で作業しました。
注）ただし、Cloud Shell 上から映像ストリームの送信を行うと、Cloud Shell のセッションが切れた際にストリームが停止してしまうので、Compute Engine で作業された方が望ましいです。

アプリケーションの作成

コンソールで Vertex AI Vision - Applications の画面を開き、Create をクリックし、アプリケーション名、リージョン、課金設定を選択します。
※ 12/16 土現時点では、us-central1 と europe-west4 しかサポートされていません…涙

映像の登録

下図のような Studio の画面が表示され、初期では Universal Input コンポーネントが表示されているので、選択して、右側ペイン内の Select input sources をクリックします。

Select processing mode では、Streaming を選択（今回はリアルタイム映像解析のため）、

Sources では Register new streams を選択し、映像ソースの名称を入力し、SUBMIT をクリックします。今回の映像ソース名は、regi-camera にしました。

映像の入力

映像の入力には、事前準備で利用した vaictl ツールを使います。
vaictl でサポートしている映像ストリームの条件はこちらに記載があります。一般的なカメラだと問題ないかと思います。

コーデック：H.264
解像度：1080p 以下
フレームレート：25 FPS以下

一点注意点としては、vaictl が取り込めるリアルタイム映像プロトコルは RTSP(Real Time Streaming Protocol)のみになります。一般的なネットワーク監視カメラだと RTSP 出力対応していることが多いですが、そうでないカメラだと ffmpeg 等のツールでの変換が必要になります。

今回はデモですのでリアルタイム映像ではなく前述の「店舗内レジの待機列のサンプル映像（mp4）」を使って vaictl でリアルタイム映像ストリームに疑似変換し、 Vision Studio に取り込んでいきます。
Ingest Videos の手順に従って作業します。

Cloud Shell 環境上で、事前にダウンロードしておいた mp4 ファイルを下記コマンドで Vision Studio に送信します。短尺映像なのでループ再生するようにオプション指定しています。
大文字の部分は皆様の環境に合わせて変更して下さい。

vaictl -p PROJECT_ID \
         -l REGION \
         -c application-cluster-0 \
         --service-endpoint visionai.googleapis.com \
send video-file to streams STREAM_NAME --file-path VIDEO_FILE_NAME --loop

うまく取り込めると、ブラウザの Streams 画面にて、当該ストリームでプレビュー表示されます（初回表示まで 30 秒程度かかるときもあります）。

映像が確認できたら、後続の処理を追加していきます。

Person Blur ぼかし処理

これは必須ではないですが、後段でアーカイブとして映像を録画保管したいので、個人を特定できないように人に「ぼかし処理」をかけようと思います。
「Universal input」コンポーネントをクリックし、Output から「Pre-trained models」→「Person Blur」を選択します。

「Person Blur」コンポーネントをクリックすると、人の場合、下記の組み合わせ 4 パターンから処理の内容を選択できます。

ぼかし処理を行う範囲（全身 or 顔のみ）
ぼかし処理の内容（黒く塗りつぶす or モザイク処理）
今回は、顔のみ黒く塗りつぶす処理をかけてみます。

ぼかし処理をかけた映像を保存するために、Vision AI Warehouse を output に指定します。
「Person Blur」コンポーネントをクリックし、Output から「Connectors」→「Vision AI Warehouse」を選択します。
Vision AI Warehouse は新規作成します。

Occupancy analytics　の設定

次に元映像の処理として、レジの混雑度を計測したいと思います。
「Universal input」コンポーネントの「Output」にて「ADD OUTPUT」をクリックし、「Specialized models」→「Occupancy analytics」を選択します。

「Occupancy analytics」コンポーネントの Settings ではデフォルトで People と Vehicles にチェックが入っているので、Vehicles のチェックを外します。

レジ待ち指定エリア設定

次は Occupancy analytics　で映像の中で人の数を計測したいエリア（レジ待ち行列エリア）を指定します。「CREATE ACTIVE ZONES/LINES」をクリックし、カメラ映像内のエリアを指定します（四角で囲みます）。
今回は、下図左エリアを serving-zone（商品提供待ち列）、右エリアを queue-zone （レジ待ち列）と２つ指定しました。

BigQuery テーブル作成

次は Occupancy analytics　で計測したそれぞれのエリアごとの待機人数を BigQuery に保存する設定を行います。

今回、BigQuery には時刻ごとの商品受け取りおよびレジの待機人数を保存しようと思います。
まず事前に BigQuery 画面にて、下記カラム構成でテーブルを作成してください。

[
  {
    "name": "ingestion_time",
    "type": "TIMESTAMP",
    "mode": "REQUIRED"
  },
  {
    "name": "serving_person_count",
    "type": "INTEGER",
    "mode": "NULLABLE"
  },
      {
    "name": "queue_person_count",
    "type": "INTEGER",
    "mode": "NULLABLE"
  },
]

BigQuery への出力および通知連携用の Cloud Functions 作成

Vision Studio で構築したアプリから通知を受けるアプリ兼 BigQuery へのデータ保存用のアプリとして Cloud Functions を作っておきます。今回は Occupancy analytics で事前定義済みのモデル Person/vehicle detector（人・車の検出）を使います。このモデルが出力するフォーマットは Model output に記載があります。処理しているフレームごとの検出数を出力してくれるようですね。
このデータを Cloud Logging に出力しつつ、BigQuery にも保存できる関数にします。

なお、単純に計測結果を生データとして BigQuery へ連携するだけであれば（通知アプリが不要であれば）下記の Cloud Functions のコードは一切必要なく、Vision Studio をポチポチするだけで、BigQuery へデータ連携することが可能です。
今回はちょっと応用っぽくやっています。

Cloud Functions の関数は、Vision Studio と同じリージョンである必要がありますのでご注意下さい。今回だと us-central1 です。
サンプルコードは下記です。Python 3.11 で「regi-queue-alert」という名称でデプロイしました。

まず BigQuery へ保存するためのスキーマ用の Protobuf ファイルを作成します。test_table_schema.proto

syntax = "proto3";

package visionai.testing;

message TestTableSchema {
  int64 ingestion_time = 1;
  int32 serving_person_count = 2;
  int32 queue_person_count = 3;
}

下記コマンドでコンパイルすると、

protoc -I=./ --python_out=./ ./test_table_schema.proto

test_table_schema_pb2.py が生成されるので、Cloud Functions にアップロードします。

requirements.txt には下記を記載。

functions-framework==3.*
click==8.1.7
cloudevents==1.10.1
deprecation==2.1.0
Flask==2.3.2
gunicorn==21.2.0
itsdangerous==2.1.2
Jinja2==3.1.2
MarkupSafe==2.1.3
pathtools3==0.2.1
watchdog==3.0.0
Werkzeug==3.0.1
protobuf==3.20.0
setuptools==69.0.2
numpy<1.26.2

main.py には下記を記載してデプロイを行います。
BigQuery へ連携するためには、レスポンスとして AppendRowsRequest をセットしないといけません。
このあたりの注意事項については Cloud Functions integration をご参照下さい。

import base64
import sys

from flask import jsonify
import functions_framework
from google.protobuf import descriptor_pb2
from google.protobuf.json_format import MessageToDict
import test_table_schema_pb2
import json

def table_schema():
  schema = descriptor_pb2.DescriptorProto()
  test_table_schema_pb2.DESCRIPTOR.message_types_by_name['TestTableSchema'].CopyToProto(schema)
  return schema

def bigquery_append_row_request(row):
  append_row_request = {}
  append_row_request['protoRows'] = {
      'writerSchema': {
          'protoDescriptor': MessageToDict(table_schema())
      },
      'rows': {
          'serializedRows':
              base64.b64encode(row.SerializeToString()).decode('utf-8')
      }
  }
  return append_row_request

@functions_framework.http
def hello_http(request):
  request_json = request.get_json(silent=False)
  #print(json.dumps(request_json))
  annotations = []
  payloads = []

  if request_json and 'annotations' in request_json:
    for annotation_with_timestamp in request_json['annotations']:
      row = test_table_schema_pb2.TestTableSchema()
      row.serving_person_count = 0
      row.queue_person_count = 0
        
      if 'annotation' in annotation_with_timestamp:
        row.ingestion_time = int(annotation_with_timestamp['ingestionTimeMicros'])
        annotation = annotation_with_timestamp['annotation']
        activeZoneCounts = annotation['stats']['activeZoneCounts']
        for activeZoneCount in activeZoneCounts:
          if activeZoneCount['annotation']['displayName'] == 'serving-zone':
            row.serving_person_count = activeZoneCount['counts'][0]['count']
            print("serving-zone : " + str(activeZoneCount['counts'][0]['count'])  + " people on the serving queue.")
          if activeZoneCount['annotation']['displayName'] == 'queue-zone':
            row.queue_person_count = activeZoneCount['counts'][0]['count']
            print("queue-zone : " + str(activeZoneCount['counts'][0]['count'])  + " people on the queue queue.")
      payloads.append(bigquery_append_row_request(row))
  for payload in payloads:
    annotations.append({'annotation': payload})
  return jsonify(annotations=annotations)

BigQuery への計測結果保存

そして次に Vision Studio のアプリ画面で、「BigQuery」コンポーネントの BigQuery Path で作成したテーブルを選択します。
そして下図のように、Occupancy analytics だけにチェックを付け、「Customize with Cloud Functions (Advanced)」を有効化し、前ステップで作成した Cloud Functions を指定します。

デプロイ

一通りの設定が完了すると、下記のような構成になっていると思います。

画面上の DEPLOY をクリックしてデプロイを行います。下図のようにサイドパネルが表示されますが、そのまま DEPLOY をクリックします。

動作確認

通知（Cloud Logging）

まず Cloud Logging を確認すると、意図した通り、出力がされています。
今回はあくまで確認用としてログ出力しただけですので、実運用としてアプリへの PUSH 通知等を行いたい場合は、この Cloud Functions から連携すれば良いことがわかります。

分析用DWH(BigQuery)

そして BigQuery にも想定通りのスキーマで保存されていました。

映像アーカイブ(Vision Warehouse)

そして Vision Warehouse にも黒塗り処理の映像として格納されていました。

料金

気になる料金は、詳細はVertex AI Vision の料金をご参照下さい。
基本的には、

従量課金（PAYG）
一部モデルのみ月額固定 + PAYG
の 2 種類です。

費用の内訳は基本的に、「アプリの費用 = ストリーミングの費用 + 分析の費用」となります。
例えば今回の例「映像を入力し、待機人数の計測」だと、以下のようになります。
（試算を単純化するためコアのシナリオのみに絞っています。Vision Warehouse と BigQuery, Cloud Functions, Cloud Logging, ぼかし処理の費用は含めておりません）

【前提】

入力映像の容量 : 60MB/分（ビットレートにも依存するので仮の数値）
処理時間 : 60分(1時間)

【単価】

入力映像単価 : $0.0085/GB(1024MB)
人数計測 : $0.10/分

【計算】

入力映像コスト : 60MB/分 x 60分 ÷ 1024MB/GB x $0.0085/GB = $0.03($0.0298828125)
分析コスト : 60分 x $0.10/分 = $6
合計 : $0.03 + $6 = $6.03/時間

分析コストが PAYG だと結構高いので、本番化する際は月額課金「$10 per stream per month」がお得ですね。（というか PAYG が $6/時間で、月額課金だと $10 って価格崩壊してる気がするが…）

まとめ

今回は通知連携用に Cloud Functions でコードを記載しましたが、それが不要であれば、以下の機能を実装したリアルタイム映像解析アプリが Vision Studio でコードを一切書く必要なく、ブラウザ上でポチポチで実現できます。

映像ストリームの受付
事前学習済み機械学習モデルの利用（計測、映像処理）
BigQuery への計測結果連携
Vision Warehouse への映像データ保存

しかも月額課金にすると安い！
今回は取り扱いませんでしたが、動画や画像ファイルを処理するアプリももちろん作ることができます！

AI を活用したリアルタイム映像解析は非常にニーズが高いもののまだまだ実社会で実装されていない領域でもありますので、ぜひ Vertex AI Vision Studio を活用して見て下さい！

Google Cloud JapanPublication

Google Cloud Japan のエンジニアを中心に情報発信をしている Publication です。Google Cloud サービスの技術情報を中心に公開しています。各記事の内容は個人の意見であり、企業を代表するものではございません。