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

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

Video AI(Video Intelligence API)とは？

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

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

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

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

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

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

1. プロジェクトの作成

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

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

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

3. Cloud Video Intelligence APIの有効化

APIの有効化ページの表示
APIの有効化ページで、ステップ1で選択または作成したプロジェクトに対する変更であることを確認して「次へ」をクリック

もしプロジェクトが異なる場合は、

上記画像の「プロジェクトの選択」の部分が異なるプロジェクトになっているので正しいプロジェクトを選択した上で、 APIの有効化ページを表示してください
請求先アカウント
「課金を有効化」をクリックして課金されるアカウントの設定を行ってください
APIを有効化
「有効にする」をクリックしてAPIの有効化の完了です

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

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

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

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

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()