背景

最近、とある事情によりGoogle Speech-to-Textで非同期の音声認識をすることになりました。Google Speech-to-Textの非同期音声認識には次のような特徴があります。

60秒を超える音声の文字起こしが可能
Google Cloud Storage（GCS）上にwavもしくはflac形式で音声ファイルを配置する必要がある
GCS上の音声ファイルは公開されている必要がある

詳しくは公式ドキュメントを参照してください。

今回、下記のような課題がありました。

音声認識させたい音声ファイルはm4a形式なので変換が必要
- FFmpegを使えば任意の音声ファイルをwav形式に変換できる
できればサーバーレスにしたい
- AWS Lambdaには先行事例がある
- LambdaからGCSにファイルをアップロードするのが面倒
- Google App Engine（GAE）はGCSと相性が良いがFFmpegを使った先行事例が見つからない

最初はAWS Lambdaでやろうとしましたが、GCSにアップロードするのが予想以上に面倒で断念しました。結局、GAEで試行錯誤しながら実装しました。終わってみればとても簡単でしたが、手探りで実装していくのはとても大変だったので、似たような課題を持った人が無駄な時間を費やさないためにこの記事を書いています。

GAEでFFmpegを利用するための準備

GAEにはスタンダード環境とフレキシブル環境という2種類の環境があります。スタンダード環境はGoogleが用意した環境で、細かいカスタマイズはできない代わりに無料枠があり、従量課金も安いという特徴があります。フレキシブル環境は任意のDockerコンテナを動かせるので何でもできますが、無料枠はなくスタンダード環境に比べると高いです。GAEで何か実装する場合は、まずスタンダード環境で動かすことを検討して、どうしてもだめな場合はフレキシブル環境を使うというのが定石です。

AWS Lambdaにはカスタムレイヤーという機能があって、LambdaでFFmpegを使う場合は自分でカスタムレイヤーを作って設定する必要があります。GAEでも似たような方法が使えないかと考え、GAEのスタンダード環境にzipで固めたFFmpegをアップロードして使う方法をがんばって調べました。

なかなかうまく行かずにいろいろな方法を試していたのですが、実は何もしなくてもGAEのスタンダード環境ではFFmpegが使えることが判明しました。^[1]

数時間の試行錯誤のあとだったので声が出ました。ここからは早かったです。

ソースコードと解説

ソースコードはこちらです。

main.py

from flask import Flask, jsonify, request
from google.cloud import storage
import os
import shlex
import subprocess
import uuid

app = Flask(__name__)

@app.route('/', methods=['POST'])
def root():
    """GCSにアップロードされた任意の音声ファイルをWAVE形式に変換してGCSにアップロードする。

    アップロードした音声ファイルのオブジェクト名を'object_name'として渡すと変換後のWAVEオブジェクトのURIが'gs_uri'として返される。
    Google Speech-to-text APIを利用するための変換処理。
    音声ファイルはm4a形式を想定しているが、ffmpegが対応していればどんな形式でも変換可能。
    GCSのバケット名はapp.yamlのenv_variablesで指定すること。
    """

    # リクエストパラメーターの解析
    request_json = request.get_json()
    object_name = request_json['object_name']

    # ファイル名
    uuid_str = str(uuid.uuid4())
    input_file_name = uuid_str + '.m4a'
    output_file_name = uuid_str + '.wav'
    input_path = '/tmp/' + input_file_name
    output_path = '/tmp/' + output_file_name

    # GCSから音声ファイルを取得
    bucket_name = os.getenv('BUCKET_NAME')
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blob_m4a = bucket.blob(object_name)
    blob_m4a.download_to_filename(input_path)

    # 音声ファイルをwavに変換
    ffmpeg_cmd = 'ffmpeg -i ' + input_path + ' -ac 1 -ar 16000 -f wav ' + output_path
    command1 = shlex.split(ffmpeg_cmd)
    subprocess.run(command1)

    # wavファイルをGCSにアップロードして公開
    blob_wav = bucket.blob(output_file_name)
    blob_wav.upload_from_filename(output_path)
    blob_wav.make_public()
    gs_uri = 'gs://' + bucket_name + '/' + output_file_name

    # 一時ファイルを削除
    os.remove(input_path)
    os.remove(output_path)

    return jsonify({
        "gs_uri": gs_uri
    })

if __name__ == '__main__':
    app.run(host='127.0.0.1', port=8080, debug=True)

単純なプログラムですが一応解説します。

リクエストパラメーターの解析

    request_json = request.get_json()
    object_name = request_json['object_name']

このAPIはリクエストにGCSのオブジェクト名を渡します。wavファイルに変換したい音声ファイルは事前にGCSにアップロードしておきます。

ファイル名とファイルパスの生成

    uuid_str = str(uuid.uuid4())
    input_file_name = uuid_str + '.m4a'
    output_file_name = uuid_str + '.wav'
    input_path = '/tmp/' + input_file_name
    output_path = '/tmp/' + output_file_name

GAEのファイルシステムは基本的にリードオンリーですが、/tmpだけは書き込むことができます。変換元の音声ファイルと変換後のwavファイルのファイルパスをここで生成しています。リクエストパラメーターのobject_nameを使わずUUIDを生成しているのは同名のファイルが存在した場合の事故防止のためです。今回の作りならobject_nameを使っても多分大丈夫ですが、こういう細かい気遣いは重要です。

GCSから音声ファイルをダウンロード

    bucket_name = os.getenv('BUCKET_NAME')
    client = storage.Client()
    bucket = client.get_bucket(bucket_name)
    blob_m4a = bucket.blob(object_name)
    blob_m4a.download_to_filename(input_path)

GCSのバケット名はapp.yamlに記載しています。

FFmpegの実行

    ffmpeg_cmd = 'ffmpeg -i ' + input_path + ' -ac 1 -ar 16000 -f wav ' + output_path
    command1 = shlex.split(ffmpeg_cmd)
    subprocess.run(command1)

Pythonから任意のコマンドを実行できるsubprocessを用いてffmpegを呼び出しています。引数をよく見ると変換後のwavは指定されていますが、変換前の形式はどこにもありません。FFmpegはファイル形式を自動で識別してくれるので指定する必要がないのです。このプログラムでは拡張子をm4aに設定していますが、実はほかの形式のファイルをアップロードしても動作します。

wavファイルをGCSにアップロードして公開

    blob_wav = bucket.blob(output_file_name)
    blob_wav.upload_from_filename(output_path)
    blob_wav.make_public()
    gs_uri = 'gs://' + bucket_name + '/' + output_file_name

変換後のwavファイルをGCSにアップロードしたあとmake_publicで公開しています。Google Speech-to-Textに音声ファイルを読み込ませるためには公開する必要があるための処置です。 何も考えずにコピペするとセキュリティ事故を起こすので注意しましょう。 私はGCS側の設定で1日後に自動で消去されるようにしています。

一時ファイルを削除

    os.remove(input_path)
    os.remove(output_path)

GAEのインスタンスは一定時間アクセスがないと自動で消えますが、アクセスがあれば再利用されます。通常は自動で消えるので忘れがちですが、使ったリソースは必ず開放するようにしましょう。忘れると思わぬ事故の原因になります。

wavファイルのURIを返して終了

    return jsonify({
        "gs_uri": gs_uri
    })

GCSにアップロードしたwavファイルのURIを返しています。このURIはそのままSpeech-to-Text APIのリクエストパラメーターに利用できます。

まとめ

GAEのスタンダード環境でFFmpegを使う方法（というか何もしなくても使えるという情報）をお伝えしました。私のように何時間も試行錯誤する人が一人でも減ることを祈っています。

脚注

https://cloud.google.com/appengine/docs/standard/python3/reference/system-packages ↩︎

Discussion