Zenn
📹

AI動画解析サービスを試してみる 2

2024/10/03に公開
Python
JSON
Google
amazonrekognition
videointelligence
tech

前回のAmazon Rekognition Videoに引き続き、AI動画解析サービスを試してみるの2回目です。

今回は GoogleのVideo AI(Video Intelligence API)を使うまでの設定とPythonスクリプトを使用した解析を実行して結果を取得し、JSONで保存してみます。

Video AI(Video Intelligence API)とは？

詳しくは公式Webサイトを見ていただくのが一番ですが、簡単に言えばローカルにあるファイルやGoogle Cloud Storageに保存してある画像や動画に対して顔検出、テキスト検出、ラベル検出などの解析が行えるサービスです。

検出について細かいカスタマイズはできませんが、手元に解析するハード・ソフト含めた環境や知識がない場合でも手軽にAI動画解析を試すことができます。

https://cloud.google.com/video-intelligence

アカウント設定の手順について

基本的には下記のURLの手順に沿って進めれば準備が完了します。特に難しいところはありませんでしたが、無料クレジットを使用するのにもクレジットカードの登録が必須のようでした。
Pythonのスクリプトで解析の実行と結果の取得をしたいのでPython用のSDKが動くところまでの設定をします。

アカウント設定の参考ページ：
https://cloud.google.com/video-intelligence/docs/common/auth?hl=ja

上記手順の途中に追加でCloud Storageに保存したファイルを読み取る権限の設定を行います。

1. プロジェクトの作成

  • プロジェクト セレクタページ から使用できるプロジェクトがある場合は既存のプロジェクトを選択し、なければ「プロジェクトの選択」をクリックし新しいプロジェクトを作成します
    ユーザー作成画面の画像1
  • 次に 「新しいプロジェクト」をクリック
    ユーザー作成画面の画像2
  • プロジェクト名を入力します
    今回は「ai-test」というユーザー名で進めます
    名前を入力したら「作成」をクリックしてプロジェクトを作成します。
    ユーザー作成画面の画像3

2. Google Cloundプロジェクトの課金設定の確認

これは次のステップにまとめて確認できます。

3. Cloud Video Intelligence APIの有効化

  • APIの有効化ページの表示
    APIの有効化ページで、ステップ1で選択または作成したプロジェクトに対する変更であることを確認して「次へ」をクリック
    API有効化の画像1
    もしプロジェクトが異なる場合は、
    API有効化の画像2
    上記画像の「プロジェクトの選択」の部分が異なるプロジェクトになっているので正しいプロジェクトを選択した上で、 APIの有効化ページを表示してください
  • 請求先アカウント
    「課金を有効化」をクリックして課金されるアカウントの設定を行ってください
    API有効化の画像3
  • APIを有効化
    「有効にする」をクリックしてAPIの有効化の完了です
    API有効化の画像4

4. サービスアカウントの作成

  • サービスアカウントを作成するプロジェクトの選択
    サービスアカウントの作成ページで、ステップ1で選択または作成したプロジェクトをクリック
    サービスアカウントの作成画像1
  • サービスアカウントを作成
    「サービス アカウント名」にアカウント名を入力します。
    サービスアカウント名も「ai-test」として進めます
    「サービス アカウントID」は自動で入力されますので、それをそのまま使用します
    「作成して続行」をクリックして次のステップに進みます
    サービスアカウントの作成画像2
  • サービスアカウントにCloud Storageの権限を追加
    • 「ロールを選択」をクリックします
      サービスアカウントの作成画像3
    • 「フィルタ」に Storage を入力し 「Storageオブジェクト閲覧者」を選択します
      サービスアカウントの作成画像4
    • 「Storage オブジェクト閲覧者」が選択されていることを確認し
      「続行」をクリックします
      サービスアカウントの作成画像5
    • 「完了」をクリックしてサービスアカウントの作成を完了します
      サービスアカウントの作成画像6

5. サービスアカウントのキーを作成します

  • サービスアカウントの一覧からサービスアカウントの「操作」カラムの縦点線のメニューをクリックし「鍵を管理」をクリック
    サービスアカウントの作成画像7
  • 「鍵を追加」をクリック
    サービスアカウントの作成画像8
  • 「キーのタイプ」で「JSON」を選択し「作成」をクリック
    サービスアカウントの作成画像9
  • 次の画面でキーがダウンロードされます
    サービスアカウントの作成画像10

6. 環境変数の設定

  • 各プラットフォームで「GOOGLE_APPLICATION_CREDENTIALS」という環境変数にステップ5で作成したJSONのパスを指定してください
    • Linux の場合
      export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-file.json"
    • Windows コマンドプロンプトの場合
      set GOOGLE_APPLICATION_CREDENTIALS=C:\path\to\service-account-file.json
    • Windows PowerShellの場合
      $env:GOOGLE_APPLICATION_CREDENTIALS="C:\path\to\service-account-file.json"

7. Google Cloud CLIのインストール

  • Google Cloud CLIのページで各プラットフォーム用のバイナリをインストールしてください

8. Google Cloud CLIの初期化

下記のコマンドを実行し初期化を行います

gcloud init

ラベル解析の実行

Cloud Storageにアップロードした動画ファイルに対して、ラベル解析を行うPythonスクリプトについて解説します。

前回行ったAmazon Rekognition Videoのラベル解析内で行っていたオブジェクトの認識と同等の解析は、Video Inteligence APIではオブジェクト追跡(オブジェクトトラッキング)となります。

Cloud Storageに動画ファイルのアップロード

事前にCloud Storageにバケットを作成し、解析に使用する動画をアップロードしておきます。

参考: https://cloud.google.com/storage/docs/discover-object-storage-console?hl=ja

Python スクリプト

ここでは複数ある解析の種類からラベルの検出をして、その結果をJSONとしてファイルに保存します。

Video Intelligence APIで動画の解析の結果は、Protbuf形式となるためPythonの辞書形式を経由してJSONに変換しています。
すべての結果は、1つのファイルにすべて保存されます。

JSONで保存しているのは、今後のスクリプトでJSONであった方が都合が良いためなので本来はどんな形式でも構いません。

設定項目

スクリプトファイルの下部にあるmain関数にユーザーと解析対象となる動画ファイルに関する設定項目があります。

  • <uri>
    Cloud Storageに保存した動画の gs://～ から始まるgsutil URIを指定します

出力結果 (例)

$ ./run-detection.py
$ ./video-ai.py
Start object detection.
.........................
ai-test.mp4-object-all.json
Finished processing.

スクリプト全体

import sys
import asyncio
import json
from google.cloud import videointelligence_v1 as videointelligence

class VideoAI:
    def __init__(self, uri):
        self.uri = uri
        self.video = self.SimpleBasename(self.uri)
        self.jobtype = 'object'
        self.model = 'builtin/latest'

    def ReturnNoneObj(x):
        return None

    def SimpleBasename(self, path):
        i = path.rfind('/') + 1
        return path[i:]

    def WriteJsonFile(self, name, data):
        print(name)
        with open(name, mode='wt', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

    async def StartDetection(self):
        dotLine = 0
        task = asyncio.create_task(self.RunDetection())
        while task.done() == False:
            await asyncio.sleep(3)
            if dotLine < 40:
                print('.', end='')
                dotLine = dotLine + 1
            else:
                print()
                dotLine = 0
            sys.stdout.flush()
            continue
        print()

        result = await task
        json_name = '{}-{}-all.json'.format(self.video, self.jobtype)
        self.WriteJsonFile(json_name, result)

        print('Finished processing.\n')

    async def RunDetection(self):
        client = videointelligence.VideoIntelligenceServiceAsyncClient()

        context = videointelligence.types.VideoContext(
                object_tracking_config = videointelligence.types.ObjectTrackingConfig(model=self.model),
                )

        operation = await client.annotate_video(
            request={
                'features': [videointelligence.Feature.OBJECT_TRACKING],
                'input_uri': self.uri,
                'video_context': context,
            }
        )

        print('Start {} detection.'.format(self.jobtype))
        result = await operation.result()

        dictdata = type(result).to_dict(result)

        return dictdata

def main():
    uri = '<cloud-storage-uri>'
    uri = 'gs://db-test-videofiles/ar-sample.mp4'

    analyzer = VideoAI(uri)
    asyncio.run(analyzer.StartDetection())

if __name__ == '__main__':
    main()

Discussion